From 2c61f63608d0e2e8a8379b6affa2b9fe29edcb14 Mon Sep 17 00:00:00 2001 From: Richard Date: Tue, 27 Oct 2026 12:46:38 +0000 Subject: [PATCH] refactor: clarify submission and routed order planning flow --- synpath/engine/engine.py | 185 +++++++++++++++++++++++++------- synpath/engine/orders/base.py | 1 + synpath/engine/orders/routed.py | 89 ++++++++++++--- 3 files changed, 223 insertions(+), 52 deletions(-) diff --git a/synpath/engine/engine.py b/synpath/engine/engine.py index b0d518e..d17d128 100644 --- a/synpath/engine/engine.py +++ b/synpath/engine/engine.py @@ -308,65 +308,139 @@ async def submit(self, request: OrderRequest, *, venue: str | None = None, risk_ `risk_reference_price` overrides the book midpoint for the limit-price risk check. """ + request = await self._prepare_submission_request(request) + # Bucket parents capture and authorize their cross-venue execution scope separately. + if is_bucket_id(request.market_id): + if venue is not None: + raise BadRequest("a bucket parent has no singular execution venue") + return await self._submit_bucket(request, risk_reference_price=risk_reference_price, + authorize_execution_accounts=authorize_execution_accounts) + return await self._submit_with_retry_protection( + request, venue=venue, risk_reference_price=risk_reference_price, precision=precision, + ) + + async def _prepare_submission_request(self, request: OrderRequest) -> OrderRequest: + """Check execution ownership and entry-point rules, then assign the worker's user.""" + # A caller cannot submit as another user. if request.user_id not in (None, self.config.user_id): raise BadRequest("user_id is assigned by the execution worker") + # Only the active lease holder may submit. if self.config.require_lease and not await self.journal.holds_lease(): raise LeaseLost("this execution worker no longer holds the journal lease") + # Persist the configured owner on the instruction. request = request.model_copy(update={"user_id": self.config.user_id}) + # RFQ orders require the quote acceptance/confirmation workflow. if request.type in (OrderType.RFQ_TAKER, OrderType.RFQ_MAKER): raise BadRequest(f"a {request.type.value} order comes from a quote: use accept_quote or confirm_quote") - if is_bucket_id(request.market_id): - if venue is not None: - raise BadRequest("a bucket parent has no singular execution venue") - return await self._submit_bucket(request, risk_reference_price=risk_reference_price, - authorize_execution_accounts=authorize_execution_accounts) + return request + + async def _submit_with_retry_protection( + self, request: OrderRequest, *, venue: str | None, + risk_reference_price: Decimal | None, precision: Precision | None, + ) -> Order: + """Reuse a durable submission or reserve its identity before executing it. + + Venue orders and stops use this retry reservation. Other managed types + retain their own lifecycle path through _submit_once(). Bucket parents + are dispatched separately by submit(). + """ + # Other managed types keep their existing lifecycle path. supported = VENUE_ORDER_TYPES | {OrderType.STOP_MARKET, OrderType.STOP_LIMIT} if request.type not in supported or request.params.get("walk"): return await self._submit_once(request, venue=venue, risk_reference_price=risk_reference_price, precision=precision) - venue = venue or self._venue_of(request) - account = request.account or self.accounts.get(venue) or Account(venue=venue) - # Caller-assigned IDs bypass the accidental-duplicate check, not same-ID retry protection. - caller_supplied_order_id = request.client_order_id is not None - request = request.model_copy(update={"account": account, - "client_order_id": request.client_order_id or f"sp-{uuid.uuid4().hex[:20]}"}) + + # Establish the account and stable order ID before looking for earlier attempts. + venue, account, request, caller_supplied_order_id = self._prepare_submission_identity(request, venue=venue) + + # Serialize lookup, reservation and execution for this submission ID. async with self._lock(f"submission:{request.client_order_id}"): - previous = await self.journal.submission(request.client_order_id) - if previous is not None: - if previous["request"] != request_identity(request): - raise BadRequest("client_order_id already identifies a different instruction or account") - result = await self._existing_submission_result(previous) - if result is not None: - return result - request = OrderRequest.stored(previous["effective_request"]) - else: - effective = as_gtd(request, now_s=self.clock(), timezone_name=self.config.session_timezone, - session_end=self.config.session_end) - prior_intent = await self.journal.intent(request.client_order_id) - if prior_intent is not None: - prior_request = OrderRequest.stored(prior_intent.request) - if prior_intent.operation != "create" or request_identity(prior_request) != request_identity(effective): - raise BadRequest("client_order_id already identifies a different durable action") - elif await self.journal.order_by_client_id(request.client_order_id) is not None: - raise BadRequest("client_order_id belongs to an older order without a verifiable instruction") - await self.journal.record_submission(request, effective, venue=venue, account=account) - if prior_intent is not None: - result = await self._existing_submission_result(await self.journal.submission(request.client_order_id)) - if result is not None: - return result - request = effective + request, previous_result = await self._resume_or_reserve_submission(request, venue=venue, account=account) + if previous_result is not None: + return previous_result + + # Execute only after confirming this instruction can safely proceed. try: return await self._submit_once( request, venue=venue, risk_reference_price=risk_reference_price, caller_supplied_order_id=caller_supplied_order_id, precision=precision, ) except (RiskRejected, BadRequest, OrderRejected) as error: - intent = await self.journal.intent(request.client_order_id) - if intent is not None and intent.state == IntentState.PLANNED: - await self.journal.mark_intent(request.client_order_id, IntentState.REJECTED, detail=str(error)) - await self.journal.reject_submission(request.client_order_id, - {"kind": type(error).__name__, "message": str(error), "rule": getattr(error, "rule", None)}) + # Preserve definite failures so retries observe the same rejection. + await self._record_submission_rejection(request, error) raise + def _prepare_submission_identity( + self, request: OrderRequest, *, venue: str | None, + ) -> tuple[str, Account, OrderRequest, bool]: + """Resolve the destination and assign an order ID while retaining its origin.""" + venue = venue or self._venue_of(request) + account = request.account or self.accounts.get(venue) or Account(venue=venue) + # Caller-assigned IDs bypass the accidental-duplicate check, not same-ID retry protection. + caller_supplied_order_id = request.client_order_id is not None + request = request.model_copy(update={ + "account": account, + "client_order_id": request.client_order_id or f"sp-{uuid.uuid4().hex[:20]}", + }) + return venue, account, request, caller_supplied_order_id + + async def _resume_or_reserve_submission( + self, request: OrderRequest, *, venue: str, account: Account, + ) -> tuple[OrderRequest, Order | None]: + """Return an earlier outcome or the effective instruction safe to execute. + + The caller must hold the submission lock through any subsequent execution. + """ + previous_submission = await self.journal.submission(request.client_order_id) + if previous_submission is None: + return await self._reserve_new_submission(request, venue=venue, account=account) + + # Reusing an ID for another instruction or account must never return an unrelated order. + if previous_submission["request"] != request_identity(request): + raise BadRequest("client_order_id already identifies a different instruction or account") + previous_result = await self._existing_submission_result(previous_submission) + if previous_result is not None: + return request, previous_result + + # Keep the original effective expiry; retrying a DAY order must not extend its life. + return OrderRequest.stored(previous_submission["effective_request"]), None + + async def _reserve_new_submission( + self, request: OrderRequest, *, venue: str, account: Account, + ) -> tuple[OrderRequest, Order | None]: + """Reserve a new submission, reconciling any older intent before allowing execution.""" + # Freeze session-relative DAY expiry before saving the instruction used for execution. + effective_request = as_gtd(request, now_s=self.clock(), timezone_name=self.config.session_timezone, + session_end=self.config.session_end) + + # Older durable actions may already own this ID without a submission reservation. + prior_intent = await self.journal.intent(request.client_order_id) + if prior_intent is not None: + prior_request = OrderRequest.stored(prior_intent.request) + if prior_intent.operation != "create" or request_identity(prior_request) != request_identity(effective_request): + raise BadRequest("client_order_id already identifies a different durable action") + elif await self.journal.order_by_client_id(request.client_order_id) is not None: + raise BadRequest("client_order_id belongs to an older order without a verifiable instruction") + + await self.journal.record_submission(request, effective_request, venue=venue, account=account) + if prior_intent is not None: + # Resolve the older send before execution; an unanswered send may already be accepted. + submission = await self.journal.submission(request.client_order_id) + previous_result = await self._existing_submission_result(submission) + if previous_result is not None: + return effective_request, previous_result + return effective_request, None + + async def _record_submission_rejection( + self, request: OrderRequest, error: RiskRejected | BadRequest | OrderRejected, + ) -> None: + """Persist a definite failure without rewriting a send that may have reached the venue.""" + # Only an unsent, planned intent can be marked rejected here. + intent = await self.journal.intent(request.client_order_id) + if intent is not None and intent.state == IntentState.PLANNED: + await self.journal.mark_intent(request.client_order_id, IntentState.REJECTED, detail=str(error)) + await self.journal.reject_submission(request.client_order_id, + {"kind": type(error).__name__, "message": str(error), "rule": getattr(error, "rule", None)}) + async def _existing_submission_result(self, submission: dict[str, Any]) -> Order | None: if submission["result_id"]: parent = self.managed_order_controller.get(submission["result_id"]) @@ -435,6 +509,21 @@ async def _submit_once(self, request: OrderRequest, *, venue: str | None = None, client_order_id = request.client_order_id or f"sp-{uuid.uuid4().hex[:20]}" request = request.model_copy(update={"client_order_id": client_order_id, "account": account}) + await self._check_venue_order_risk( + request, venue=venue, account=account, risk_reference_price=risk_reference_price, + caller_supplied_order_id=caller_supplied_order_id, + ) + order = await self._send_venue_order_with_journaled_intent( + request, venue=venue, account=account, adapter=adapter, precision=precision, + ) + return await self._record_accepted_venue_order(order, request, venue=venue) + + async def _check_venue_order_risk( + self, request: OrderRequest, *, venue: str, account: Account, + risk_reference_price: Decimal | None, caller_supplied_order_id: bool, + ) -> None: + """Reject and explain a failed risk check before any venue intent is written.""" + client_order_id = request.client_order_id decision = self.evaluate_order_risk( request, venue=venue, account=account, risk_reference_price=risk_reference_price, caller_supplied_order_id=caller_supplied_order_id, @@ -447,6 +536,16 @@ async def _submit_once(self, request: OrderRequest, *, venue: str | None = None, }, key=client_order_id) raise RiskRejected(decision.message, rule=decision.rule) + async def _send_venue_order_with_journaled_intent( + self, request: OrderRequest, *, venue: str, account: Account, + adapter: TradingExchange, precision: Precision | None, + ) -> Order: + """Persist the intent before sending; leave unanswered sends for reconciliation. + + Keep the sending transition and venue call under the same dispatch lock. + A timeout cannot prove rejection, so it must never trigger a fresh send. + """ + client_order_id = request.client_order_id await self.journal.record_intent(request, client_order_id=client_order_id, venue=venue, account=account) async with self._lock(f"{venue}:{client_order_id}"): await self.journal.mark_intent(client_order_id, IntentState.SENDING, bump_attempt=True) @@ -473,6 +572,11 @@ async def _submit_once(self, request: OrderRequest, *, venue: str | None = None, await self.journal.mark_intent(client_order_id, IntentState.FAILED, detail=str(exc)) await self.bus.publish("order.failed", {"client_order_id": client_order_id, "error": str(exc)}, key=client_order_id) raise + return order + + async def _record_accepted_venue_order(self, order: Order, request: OrderRequest, *, venue: str) -> Order: + """Save the venue response before marking its intent sent and announcing acceptance.""" + client_order_id = request.client_order_id order = self._stamp(order, request) await self.journal.upsert_order(order, event="order.accepted") self._remember(order) @@ -659,6 +763,7 @@ def _child_venue(parent: Any, request: OrderRequest) -> str: return parent.venue async def cancel_child(self, parent: Any, order_id: str, venue: str | None = None) -> Order | None: + """Cancel a managed child through its controller, or a venue child through cancel().""" if order_id in self.managed_order_controller.parents: return (await self.managed_order_controller.cancel(order_id, reason=f"cancelled by {parent.id}")).as_order() child = parent.child_of(order_id, venue) diff --git a/synpath/engine/orders/base.py b/synpath/engine/orders/base.py index 0aae70f..f314b9b 100644 --- a/synpath/engine/orders/base.py +++ b/synpath/engine/orders/base.py @@ -209,6 +209,7 @@ async def submit_child(self, request: OrderRequest, **kw: Any) -> Order: return await self.engine.submit_child(self.parent, request, **kw) async def cancel_child(self, order_id: str, venue: str | None = None) -> Order | None: + """Delegate to Engine.cancel_child(), supplying this context's managed parent.""" return await self.engine.cancel_child(self.parent, order_id, venue=venue) async def publish(self, kind: str, payload: dict[str, Any] | None = None) -> None: diff --git a/synpath/engine/orders/routed.py b/synpath/engine/orders/routed.py index b1c6502..c67598a 100644 --- a/synpath/engine/orders/routed.py +++ b/synpath/engine/orders/routed.py @@ -152,9 +152,12 @@ def known_filled(self) -> Decimal: def known_remaining(self) -> Decimal: return self.remaining - def budget(self) -> Decimal: - """Size new legs may add: what is left after every live leg's whole - size is reserved and every closed leg's fill is counted.""" + def remaining_quantity_after_fills_and_outstanding_orders(self) -> Decimal: + """Quantity available for new orders after fills and live reservations. + + A live child's original quantity covers both its fills and its remaining + outstanding quantity. An ended child reserves only its filled quantity. + """ reserved = sum((c.quantity if c.live else self.known(c) for c in self.children), ZERO) return max(ZERO, self.quantity - reserved) @@ -306,30 +309,73 @@ async def stop(self, ctx: Context, reason: str) -> None: # -- the plan and the legs ------------------------------------------------ - def may_move(self, ctx: Context, leg: LegState) -> bool: + def minimum_stay_has_elapsed(self, ctx: Context, leg: LegState) -> bool: + """Whether this leg has rested long enough to allow routine cancel/replacement.""" return ctx.now - leg.placed_at >= self.min_stay_s async def replan(self, ctx: Context, *, first: bool = False) -> None: - """Re-entrant safe: a cancel inside a plan can hand the engine a - terminal child, whose `on_child_order_updated` asks for a plan of its own. That - request is noted and honoured once this plan is through.""" + """Run planning passes in sequence, remembering requests received during a pass. + + Cancelling a child can synchronously deliver an order update, whose + callback calls replan() before the current pass returns. The controller + permits these callbacks on the same task, so its per-order lock alone + does not prevent nested planning. These flags defer that work until the + current pass finishes updating the legs and quantity reservations. + """ + # A stopping or finished parent must not allocate more venue orders. if self.state != "working": return + + # The active pass owns the leg changes. Remember that another pass is + # needed, then return to it instead of recursively changing the same legs. + # Several nested requests share one flag because the next pass reads + # the latest state rather than replaying each notification separately. if self._replanning: self._replan_again = True return + + # Set the guard before awaiting anything that could call back into us. self._replanning = True try: + # Only the initial placement uses first=True; later reallocations + # are eligible to increment the parent's routing-round counter. await self._replan(ctx, first=first) + + # Honour deferred requests after the current pass settles. A pass + # may also stop the parent, in which case no further routing is allowed. while self._replan_again and self.state == "working": + # Consume this request before awaiting: a callback during the + # next pass must be able to request yet another pass. self._replan_again = False await self._replan(ctx) finally: + # Release the guard even on failure, and leave no deferred request + # attached to this invocation when a later event calls replan(). self._replanning = False self._replan_again = False async def _replan(self, ctx: Context, *, first: bool = False) -> None: - """Reconcile the new allocation with existing children before adding orders.""" + """Reconcile a fresh proposal with existing children in one planning pass. + + First remove inactive leg references and trim live children's outstanding + quantity to fit the parent quantity still needed. Then plan from available + books for target minus all known fills, initially ignoring outstanding orders. + + For each tracked child, keep it unchanged if its side and price match the + proposal and its outstanding quantity fits within the remaining target, + even when its size differs from the proposed size. Otherwise cancel it + once its minimum stay has elapsed. Keep children with no current book + because missing data alone does not justify cancellation. + + Then add orders from the remaining proposals, skipping member markets + where a child was kept. Reserve all live children's outstanding quantity + before adding orders so the total does not exceed the requested quantity. + Current leg tracking is per member market, not per venue. + + Report and save the result, or begin stopping when a routing bound is met. + Callbacks can change lifecycle state during awaits; only working parents + may continue allocating. The replan() wrapper handles deferred passes. + """ if self.state != "working": return if self.known_remaining <= 0: @@ -353,7 +399,12 @@ async def _replan(self, ctx: Context, *, first: bool = False) -> None: ) async def _clean_up_existing_legs(self, ctx: Context) -> int: - """Remove inactive legs and cancel excess resting quantity, newest first.""" + """Clean up tracked allocations before calculating a fresh proposal. + + First remove leg metadata whose child is missing or no longer live. + Then cancel excess outstanding child quantity so it fits within the + parent's remaining target. Return the number of children canceled. + """ self._forget_dead_legs() return await self._trim(ctx) @@ -400,6 +451,7 @@ async def _keep_or_cancel_existing_legs( stage cannot submit another child there during this round. """ canceled_count = 0 + # Outstanding quantity retained so far by the missing-book and matching-price checks. kept_resting_quantity = ZERO for market_id, leg in list(self.legs.items()): child = self.child_of(leg.order_id, leg.venue) @@ -422,12 +474,13 @@ async def _keep_or_cancel_existing_legs( if (proposed_leg is not None and proposed_leg.price == leg.price and proposed_leg.side == leg.side and kept_resting_quantity + resting_quantity <= self.known_remaining): kept_resting_quantity += resting_quantity + # Skip new placement on this market because its existing child is kept. proposed_legs.pop(market_id) continue # The proposal no longer fits, but the minimum stay has not passed. - # Keep the child for now; budget() still reserves its outstanding size. - if not self.may_move(ctx, leg): + # Keep the child for now; its outstanding quantity remains reserved. + if not self.minimum_stay_has_elapsed(ctx, leg): proposed_legs.pop(market_id, None) continue @@ -443,7 +496,7 @@ async def _place_new_legs(self, ctx: Context, proposed_legs: dict[str, router.Le """Submit remaining proposals within the unreserved quantity and venue minimums.""" # Reserve every live child's size before adding orders, including those # kept despite missing books or an unelapsed minimum stay. - available_quantity = self.budget() + available_quantity = self.remaining_quantity_after_fills_and_outstanding_orders() placed_count = 0 for market_id, proposed_leg in proposed_legs.items(): # A venue response can call back into the parent and start stopping. @@ -528,6 +581,18 @@ def newest_first(item: Any) -> tuple[int, int, float]: return pulled def _forget_dead_legs(self) -> None: + """Remove current-leg metadata that no longer points to a live child. + + self.legs tracks an allocation already submitted on each member market: + its child order ID, side, price, quantity and placement information. + It is separate from the fresh proposal returned by router.plan(). + + A leg is dead when its child is missing or its recorded status is no + longer live, for example after a full fill or cancellation. Remove only + the entry in self.legs; retain the child and its fills in self.children + so they still count toward parent progress. This method sends no venue + requests and does not cancel orders. + """ for market_id, leg in list(self.legs.items()): child = self.child_of(leg.order_id, leg.venue) if child is None or not child.live: