Building a kill switch

Everything in chapters 73-74 (loss limits, exposure limits) decides *whether to open new risk*. A kill switch is the separate, more severe mechanism for *stopping everything immediately* — manual or automatic — including cancelling resting orders and, depending on severity, closing open positions.

Levels of severity — don't collapse these into one action

from enum import Enum

class KillLevel(Enum):
    HALT_NEW_ENTRIES = 1     # stop opening new positions, let existing ones run their course
    CANCEL_PENDING = 2        # + cancel all resting/pending orders
    FLATTEN_ALL = 3            # + market-close every open position immediately

class KillSwitch:
    def __init__(self, kite, order_manager):
        self.kite = kite
        self.om = order_manager
        self.active_level = None

    def trigger(self, level: KillLevel, reason: str):
        logging.critical(f"KILL SWITCH triggered: {level.name} — {reason}")
        send_alert(f"KILL SWITCH: {level.name} — {reason}")
        self.active_level = level

        if level.value >= KillLevel.CANCEL_PENDING.value:
            self._cancel_all_orders()
        if level.value >= KillLevel.FLATTEN_ALL.value:
            self._flatten_all_positions()

    def is_entry_allowed(self) -> bool:
        return self.active_level is None

    def _cancel_all_orders(self):
        for o in self.kite.orders():
            if o["status"] in ("OPEN", "TRIGGER PENDING"):
                self.om.cancel(o["order_id"])

    def _flatten_all_positions(self):
        for p in self.kite.positions()["net"]:
            if p["quantity"] == 0:
                continue
            side = "SELL" if p["quantity"] > 0 else "BUY"
            self.om.place(strategy_code="kill-switch", exchange=p["exchange"],
                           tradingsymbol=p["tradingsymbol"], transaction_type=side,
                           quantity=abs(p["quantity"]), product=p["product"], order_type="MARKET")

Automatic triggers — wire these to real conditions, not just a manual button

def check_automatic_kill_conditions(kite, kill_switch: KillSwitch, config: dict):
    equity = get_current_equity(kite)
    if equity < config["start_equity"] * (1 - config["max_daily_loss_pct"] / 100):
        kill_switch.trigger(KillLevel.HALT_NEW_ENTRIES, "Daily loss limit breached")

    if price_cache.is_stale(critical_token, max_age=60):
        kill_switch.trigger(KillLevel.HALT_NEW_ENTRIES, "Market data feed stale")

    if consecutive_order_rejections(order_log, window_minutes=5) > 5:
        kill_switch.trigger(KillLevel.CANCEL_PENDING, "Repeated order rejections — possible bug")

Manual kill switch — needs to be reachable even if your main bot process is unresponsive

# A tiny separate script, runnable independently of the main bot process
def emergency_flatten():
    kite = get_authenticated_kite()   # chapter 10
    om = OrderManager(kite)
    ks = KillSwitch(kite, om)
    ks.trigger(KillLevel.FLATTEN_ALL, "Manual emergency stop")

if __name__ == "__main__":
    emergency_flatten()

Keep this as a standalone script you can run from a phone via SSH or a simple webhook (chapter 91's Telegram bot can trigger this) — a kill switch that only works if your main bot's event loop is healthy defeats its own purpose in exactly the scenario where you need it most (the bot hung or misbehaving).

Next: 091 — Alerting via Telegram