Skip to content

fix: stop order cycle after unknown broker outcome #461

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Pigbibi merged 1 commit into from
Sep 7, 2026
+250 −71
Merged

fix: stop order cycle after unknown broker outcome #461

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
93 changes: 53 additions & 40 deletions application/execution_service.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1043,6 +1043,7 @@ def execute_rebalance_cycle(
small_account_bootstrap_note_keys: set[str] = set()
action_done = False
sell_submitted = False
submission_halted = False
pending_sell_release_symbols: list[str] = []
threshold_value = float(execution["trade_threshold_value"])
limit_order_symbols = set(
Expand Down Expand Up @@ -1206,7 +1207,16 @@ def append_order_id_suffix(log_message, order_id):
suffix = f"[order_id={order_id_text}]"
return f"{log_message} {suffix}"

def notify_submission_issue(title, detail):
try:
notify_issue(title, detail)
except Exception:
print("order_notification_failed", flush=True)

def submit_order_via_port(symbol, order_type, side, quantity, log_message, *, submitted_price=None):
nonlocal submission_halted
if submission_halted:
return False
if fractional_buy_execution and side == "buy":
normalized_quantity = _normalize_buy_quantity(
quantity,
Expand Down Expand Up @@ -1235,59 +1245,52 @@ def submit_order_via_port(symbol, order_type, side, quantity, log_message, *, su
try:
report = execution_port.submit_order(order_intent)
except Exception:
notify_issue(
"Order submit failed",
(
f"Symbol: {symbol} Side: {side_text} Qty: {quantity} "
f"Type: {order_type} Price: {submitted_price if submitted_price is not None else 'MO'}\n"
f"{traceback.format_exc()}"
),
)
return False

status = str(report.status or "").strip().lower()
if status not in {"submitted", "accepted"}:
detail = report.raw_payload.get("detail", report.status) if isinstance(report.raw_payload, Mapping) else report.status
notify_issue(
"Order submit failed",
(
f"Symbol: {symbol} Side: {side_text} Qty: {quantity} "
f"Type: {order_type} Price: {submitted_price if submitted_price is not None else 'MO'}\n"
f"Status: {detail}"
),
)
report = None
status = str(getattr(report, "status", "") or "").strip().lower()
if status == "rejected":
notify_submission_issue("Order submit failed", "order_rejected")
return False

log_with_order_id = append_order_id_suffix(log_message, report.broker_order_id)
pending_log = translator("order_pending_confirmation", detail=log_with_order_id)
print(with_prefix(pending_log), flush=True)
logs.append(pending_log)
known_submission = status in {"submitted", "accepted", "filled", "partially_filled"}
if not known_submission:
submission_halted = True
order_payload = {
"symbol": str(symbol or "").strip().upper(),
"side": str(side or "").strip().lower(),
"quantity": float(order_intent.quantity or 0.0),
"order_type": str(order_type or "").strip().lower(),
"status": "pending_reconciliation",
"submission_status": report.status,
"status": "pending_reconciliation" if known_submission else "unknown",
"submission_status": status if known_submission else "unknown",
}
if submitted_price is not None:
order_payload["price"] = round(float(submitted_price), 4)
if order_type == "limit":
order_payload["limit_price"] = round(float(submitted_price), 4)
if report.broker_order_id:
order_payload["broker_order_id"] = report.broker_order_id
submitted_orders.append(order_payload)
broker_order_id = getattr(report, "broker_order_id", None)
if broker_order_id:
order_payload["broker_order_id"] = broker_order_id
if known_submission:
order_payload["filled_quantity"] = report.filled_quantity
if report.average_fill_price is not None:
order_payload["average_fill_price"] = report.average_fill_price
# Preserve the broker fact before notifications or other fallible post-submit work.
pending_orders.append(order_payload)
if not known_submission:
notify_submission_issue("Order submit failed", "broker_outcome_unknown_reconciliation_required")
return False
submitted_orders.append(order_payload)
if str(side or "").strip().lower() == "sell":
submitted_sell_orders.append(order_payload)
if post_submit_order is not None:
try:
try:
log_with_order_id = append_order_id_suffix(log_message, broker_order_id)
pending_log = translator("order_pending_confirmation", detail=log_with_order_id)
logs.append(pending_log)
print(with_prefix(pending_log), flush=True)
if post_submit_order is not None:
post_submit_order(trade_context, order_intent, report)
except Exception:
notify_issue(
"Order post-submit hook failed",
f"Symbol: {symbol} Side: {side_text} Qty: {quantity}\n{traceback.format_exc()}",
)
except Exception:
submission_halted = True
notify_submission_issue("Order post-submit hook failed", "post_submit_processing_failed_reconciliation_required")
return True

def record_dry_run(symbol, side, quantity, price, *, order_type):
Expand Down Expand Up @@ -1319,6 +1322,8 @@ def record_dry_run(symbol, side, quantity, price, *, order_type):
return True

for symbol in strategy_assets:
if submission_halted:
break
if _sell_delta_exceeds_floor(
current_value=market_values[symbol],
target_value=target_values[symbol],
Expand Down Expand Up @@ -1381,6 +1386,8 @@ def record_dry_run(symbol, side, quantity, price, *, order_type):
translator("market_sell", symbol=symbol, qty=quantity_text, price=round(price, 2)),
)

if submission_halted and not submitted:
break
if submitted:
action_done = True
sell_submitted = True
Expand Down Expand Up @@ -1428,7 +1435,8 @@ def record_dry_run(symbol, side, quantity, price, *, order_type):
if small_account_buy_blocked or account_new_risk_buy_blocked:
funding_buy_candidates = []
if (
not sell_submitted
not submission_halted
and not sell_submitted
and funding_buy_candidates
and cash_sweep_symbol
and sellable_quantities.get(cash_sweep_symbol, 0.0) > 0.0
Expand Down Expand Up @@ -1494,7 +1502,7 @@ def record_dry_run(symbol, side, quantity, price, *, order_type):
sell_submitted = True
cash_sweep_sold_this_cycle = True

if sell_submitted:
if sell_submitted and not submission_halted:
if dry_run_only and dry_run_sale_proceeds > 0.0:
simulated_cash = float(dry_run_sale_proceeds)
available_cash = max(0.0, available_cash + simulated_cash)
Expand Down Expand Up @@ -1611,6 +1619,8 @@ def record_dry_run(symbol, side, quantity, price, *, order_type):
if (target_values[symbol] - market_values[symbol]) > threshold_value
and abs(target_values[symbol] - market_values[symbol]) > current_min_trade
]
if submission_halted:
buy_candidates = []
buys_blocked_reason: str | None = None
if small_account_buy_blocked and buy_candidates:
buys_blocked_reason = "small_account_below_recommended_equity"
Expand Down Expand Up @@ -1690,6 +1700,8 @@ def record_dry_run(symbol, side, quantity, price, *, order_type):
buy_candidates = []

for symbol in buy_candidates:
if submission_halted:
break
diff = target_values[symbol] - market_values[symbol]
price = safe_quote_last_price(
market_symbol(symbol),
Expand Down Expand Up @@ -1837,7 +1849,8 @@ def record_dry_run(symbol, side, quantity, price, *, order_type):
allocation.get("small_account_safe_haven_cash_substituted_symbols")
)
if (
not cash_sweep_sold_this_cycle
not submission_halted
and not cash_sweep_sold_this_cycle
and cash_sweep_symbol
and cash_sweep_symbol in strategy_assets
and not small_account_buy_blocked
Expand Down
73 changes: 42 additions & 31 deletions application/rebalance_service.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
AccountIdentityPolicy,
evaluate_account_identity,
)
from quant_platform_kit.common.models import ExecutionReport
from quant_platform_kit.common.port_adapters import CallableExecutionPort
from quant_platform_kit.longbridge.market_data import fetch_lot_sizes
from application.signal_snapshot import build_signal_snapshot
Expand Down Expand Up @@ -339,7 +340,7 @@ def _should_record_execution_marker(*, result: ExecutionCycleResult, config: Lon
return False
if bool(getattr(config, "dry_run_only", False)) and tuple(getattr(result, "dry_run_orders", ()) or ()):
return True
return bool(getattr(result, "action_done", False))
return bool(getattr(result, "action_done", False) or getattr(result, "pending_orders", ()))


def _record_execution_marker(
Expand All @@ -353,7 +354,8 @@ def _record_execution_marker(
if not store or not marker_key:
return
try:
store.record_marker(
record = store.record_outcome if result.pending_orders else store.record_marker
record(
marker_key,
metadata={
"strategy_profile": getattr(config, "strategy_profile", ""),
Expand All @@ -362,15 +364,16 @@ def _record_execution_marker(
"action_done": bool(getattr(result, "action_done", False)),
"dry_run_orders_count": len(tuple(getattr(result, "dry_run_orders", ()) or ())),
"pending_orders_count": len(tuple(getattr(result, "pending_orders", ()) or ())),
"pending_orders": list(result.pending_orders),
"signal_date": str(dict(getattr(result, "execution", {}) or {}).get("signal_date") or ""),
"effective_date": str(dict(getattr(result, "execution", {}) or {}).get("effective_date") or ""),
},
)
except Exception as exc:
notify_issue(
"Execution marker write failed",
f"Marker: {marker_key}\n{type(exc).__name__}: {exc}",
)
except Exception:
try:
notify_issue("Execution marker write failed", "execution_outcome_persistence_failed")
except Exception:
print("order_notification_failed", flush=True)



Expand Down Expand Up @@ -545,6 +548,10 @@ def fetch_replanned_state():

def submit_claimed_order(order_intent):
nonlocal execution_claim_attempted, execution_claim_acquired
rejected = ExecutionReport(
symbol=order_intent.symbol, side=order_intent.side,
quantity=order_intent.quantity, status="rejected",
)
# Claim only when an order is ready; failed claims are never retried in this cycle.
if not execution_claim_attempted:
execution_claim_attempted = True
Expand All @@ -564,24 +571,25 @@ def submit_claimed_order(order_intent):
},
)
except Exception:
raise RuntimeError(
"LongBridge account owner fence unavailable; refusing broker submission"
) from None
return rejected
if not owner_claim.allowed:
raise RuntimeError(
"LongBridge account owner fence contested for "
f"account={account_id}; owner={owner_claim.owner_id!r} "
f"contested_by={owner_id!r}"
)
return rejected
try:
execution_claim_acquired = bool(execution_state_store.claim_marker(
execution_marker_key,
metadata={"platform": "longbridge", "strategy_profile": config.strategy_profile},
metadata={
"platform": "longbridge", "strategy_profile": config.strategy_profile,
"order_intent": {
"symbol": order_intent.symbol, "side": order_intent.side,
"quantity": order_intent.quantity, "order_type": order_intent.order_type,
"limit_price": order_intent.limit_price,
},
},
))
except Exception:
raise RuntimeError("LongBridge execution claim unavailable; refusing broker submission") from None
return rejected
if not execution_claim_acquired:
raise RuntimeError("LongBridge execution claim required; refusing broker submission")
return rejected
return delegate.submit_order(order_intent)

execution_port = CallableExecutionPort(submit_claimed_order)
Expand Down Expand Up @@ -678,20 +686,23 @@ def submit_claimed_order(order_intent):
pending_orders = tuple(getattr(execution_result, "pending_orders", ()) or ())

if pending_orders:
notification_publisher.publish(
notification_renderers.render_rebalance_notification(
execution=execution,
logs=logs,
skip_logs=skip_logs,
note_logs=note_logs,
translator=config.translator,
separator=config.separator,
strategy_display_name=config.strategy_display_name,
dry_run_only=config.dry_run_only,
extra_notification_lines=config.extra_notification_lines,
title_key="pending_order_title",
try:
notification_publisher.publish(
notification_renderers.render_rebalance_notification(
execution=execution,
logs=logs,
skip_logs=skip_logs,
note_logs=note_logs,
translator=config.translator,
separator=config.separator,
strategy_display_name=config.strategy_display_name,
dry_run_only=config.dry_run_only,
extra_notification_lines=config.extra_notification_lines,
title_key="pending_order_title",
)
)
)
except Exception:
print("pending_order_notification_failed", flush=True)
elif action_done:
notification_publisher.publish(
notification_renderers.render_rebalance_notification(
Expand Down
Loading