46 lines
1.7 KiB
Python
46 lines
1.7 KiB
Python
from __future__ import annotations
|
|
|
|
|
|
from .dsl import PlanDelta
|
|
|
|
|
|
class ExecutionEngine:
|
|
"""Minimal execution adapter: pretend to route orders across venues."""
|
|
|
|
def __init__(self):
|
|
pass
|
|
|
|
def route(self, plan: PlanDelta, signals=None):
|
|
"""Build a naive routing plan for each delta in the PlanDelta.
|
|
|
|
This lightweight implementation serves as a compatibility shim for
|
|
tests that expect an ExecutionEngine to expose a `route()` method.
|
|
It returns a list of human-readable route descriptions that include
|
|
the substring 'route_delta_to' so tests can validate routing was invoked.
|
|
"""
|
|
routes = []
|
|
# Normalize plan entries to a list of delta-like objects.
|
|
entries = []
|
|
if getattr(plan, "deltas", None):
|
|
entries = plan.deltas
|
|
elif getattr(plan, "delta", None):
|
|
entries = plan.delta
|
|
# Build a simple textual route per entry
|
|
for idx, entry in enumerate(entries or []):
|
|
# If the entry is a dict-like, reflect its contents; else try common attr
|
|
if isinstance(entry, dict):
|
|
delta_desc = dict(entry)
|
|
else:
|
|
delta_desc = getattr(entry, "delta_positions", {}) or {}
|
|
routes.append(f"route_delta_to venue{idx} {delta_desc}")
|
|
if not routes:
|
|
routes.append("route_delta_to: default")
|
|
return routes
|
|
|
|
def execute(self, plan: PlanDelta) -> dict:
|
|
# Naive: compute an execution cost proxy from plan
|
|
cost = max(0.0, plan.total_cost)
|
|
# pretend PnL impact as a function of plan hedges and a deterministic factor
|
|
pnl = -cost * 0.5
|
|
return {"status": "ok", "cost": cost, "pnl": pnl}
|