Building an Order Manager abstraction
Chapters 43-66 covered raw API calls. A real bot shouldn't scatter kite.place_order() calls through strategy code — wrap them behind an OrderManager that centralizes idempotency (ch 58), tagging (ch 65), retry/error handling (ch 57), and reconciliation (ch 66) in one place.
# orders/manager.py
import logging, time, uuid
from kiteconnect.exceptions import OrderException, NetworkException
class OrderManager:
def __init__(self, kite):
self.kite = kite
self.local_orders: dict[str, dict] = {} # order_id -> our record
def place(self, strategy_code: str, **params) -> dict:
tag = params.pop("tag", None) or f"{strategy_code}-{uuid.uuid4().hex[:12]}"
params["tag"] = tag
try:
order_id = self._place_with_retry(**params)
except OrderException as e:
logging.error(f"Order rejected: {e}")
return {"success": False, "order_id": None, "reason": str(e)}
self.local_orders[order_id] = {"strategy": strategy_code, "params": params, "status": "PENDING"}
return {"success": True, "order_id": order_id, "reason": None}
def _place_with_retry(self, **params, max_attempts=2):
tag = params["tag"]
for attempt in range(max_attempts):
try:
return self.kite.place_order(**params)
except NetworkException:
existing = [o for o in self.kite.orders() if o.get("tag") == tag]
if existing:
return existing[0]["order_id"]
if attempt == max_attempts - 1:
raise
time.sleep(1)
def cancel(self, order_id: str) -> bool:
order = next((o for o in self.kite.orders() if o["order_id"] == order_id), None)
if not order:
return False
try:
self.kite.cancel_order(variety=order["variety"], order_id=order_id)
return True
except OrderException:
return False
def sync_status(self):
"""Refresh local records against broker truth — call periodically."""
broker_orders = {o["order_id"]: o for o in self.kite.orders()}
for order_id, record in self.local_orders.items():
if order_id in broker_orders:
record["status"] = broker_orders[order_id]["status"]
record["filled_quantity"] = broker_orders[order_id]["filled_quantity"]
def open_orders_for(self, strategy_code: str) -> list[dict]:
return [
{**rec, "order_id": oid} for oid, rec in self.local_orders.items()
if rec["strategy"] == strategy_code and rec["status"] in ("PENDING", "OPEN", "TRIGGER PENDING")
]Why centralize this instead of leaving it in strategy code
Strategy code (chapters 79+) should express *decisions* ("enter long here," "exit here") — not retry loops, tag generation, or exception handling boilerplate. A single, well-tested OrderManager means every strategy gets idempotency and error handling for free, and a bug fix here (e.g. a new rejection reason to handle) benefits every strategy at once instead of needing to be patched into N copy-pasted order-placement blocks.