""" AkDevz AI Market Intelligence — Autonomous Execution & Position Management Engine Institutional quantitative execution engine supporting: - Paper Trading and Compliant Live Broker Execution - Mathematical risk gates and capital allocation - Idempotency keys & duplicate-order prevention - Trailing stop-loss & protective target monitoring - Broker position reconciliation & trade ledger persistence - Emergency risk kill switch """ import sys import time import json import uuid import hashlib import logging from datetime import datetime from pathlib import Path from typing import Dict, Any, List, Optional, Tuple from zoneinfo import ZoneInfo PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) from python.config.settings import settings from python.brokers.broker_failover import BrokerFailoverManager from python.brokers.fyers_adapter import FyersBrokerAdapter import pymysql logger = logging.getLogger(__name__) IST = ZoneInfo("Asia/Kolkata") class ExecutionEngine: def __init__(self): self.db_params = { "host": settings.db_host, "port": settings.db_port, "user": settings.db_username, "password": settings.db_password, "database": settings.db_database, "cursorclass": pymysql.cursors.DictCursor, "autocommit": True } def get_connection(self): return pymysql.connect(**self.db_params) def get_risk_config(self) -> Dict[str, Any]: conn = self.get_connection() cur = conn.cursor() cur.execute("SELECT * FROM risk_configuration WHERE id = 1") row = cur.fetchone() or { "live_execution_enabled": 0, "paper_trading_enabled": 1, "selected_execution_broker": "FYERS", "max_daily_loss": 15000.0, "max_loss_per_trade": 3000.0, "max_open_positions": 3, "max_order_value": 100000.0, "max_daily_trades": 10, "kill_switch_active": 0, "trailing_sl_enabled": 1, "trailing_sl_points": 15.0 } conn.close() return row def generate_idempotency_key(self, symbol: str, direction: str, strategy: str) -> str: """ Creates a deterministic time-bucketed (5-min) idempotency key to prevent duplicate order execution for the same setup. """ bucket = int(time.time() / 300) raw = f"{symbol}:{direction}:{strategy}:{bucket}" return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:32] def validate_risk_gates(self, candidate: Dict[str, Any], risk_cfg: Dict[str, Any]) -> Tuple[bool, str]: """ Evaluates mathematical safety and capital exposure limits. """ if risk_cfg.get("kill_switch_active"): return False, "EMERGENCY_KILL_SWITCH_ACTIVE: Execution is completely halted." conn = self.get_connection() cur = conn.cursor() # Check total open positions count is_paper = 0 if risk_cfg.get("live_execution_enabled") else 1 cur.execute("SELECT COUNT(*) as cnt FROM trading_positions WHERE status = 'OPEN' AND is_paper = %s", (is_paper,)) open_cnt = cur.fetchone()["cnt"] if open_cnt >= risk_cfg.get("max_open_positions", 3): conn.close() return False, f"MAX_OPEN_POSITIONS_REACHED: Currently {open_cnt} active positions." # Check total daily trades count cur.execute("SELECT COUNT(*) as cnt FROM trading_orders WHERE DATE(placed_at) = CURDATE() AND is_paper = %s", (is_paper,)) daily_trades = cur.fetchone()["cnt"] if daily_trades >= risk_cfg.get("max_daily_trades", 10): conn.close() return False, f"MAX_DAILY_TRADES_REACHED: {daily_trades} orders submitted today." # Check daily realized loss limit cur.execute("SELECT SUM(realized_pnl) as total_pnl FROM trading_positions WHERE DATE(closed_at) = CURDATE() AND is_paper = %s", (is_paper,)) row = cur.fetchone() today_pnl = float(row["total_pnl"] or 0.0) if today_pnl < -float(risk_cfg.get("max_daily_loss", 15000.0)): conn.close() return False, f"MAX_DAILY_LOSS_EXCEEDED: Loss ₹{abs(today_pnl):.2f} exceeds limit." conn.close() return True, "RISK_GATES_PASSED" def execute_order(self, candidate: Dict[str, Any]) -> Dict[str, Any]: """ Executes a validated trade setup either in Paper Trading mode (default) or Live Broker mode (if explicitly authorized by administrator). """ risk_cfg = self.get_risk_config() passed, reason = self.validate_risk_gates(candidate, risk_cfg) if not passed: return {"status": "rejected", "reason": reason} symbol = candidate.get("symbol", "") direction = candidate.get("direction", "BUY") entry_price = float(candidate.get("entry_price", 0.0)) stop_loss = float(candidate.get("stop_loss", 0.0)) target_1 = float(candidate.get("target_1", 0.0)) target_2 = float(candidate.get("target_2", 0.0)) quantity = int(candidate.get("quantity", 1)) strategy = candidate.get("strategy_name", "QUANT_ALPHA") idempotency_key = self.generate_idempotency_key(symbol, direction, strategy) is_paper = 0 if risk_cfg.get("live_execution_enabled") else 1 provider_code = risk_cfg.get("selected_execution_broker", "FYERS") conn = self.get_connection() cur = conn.cursor() # Idempotency check: Reject duplicate order within current timeframe window cur.execute("SELECT id, status FROM trading_orders WHERE idempotency_key = %s", (idempotency_key,)) existing = cur.fetchone() if existing: conn.close() return { "status": "duplicate_prevented", "message": f"Order already exists with idempotency key {idempotency_key}." } # Handle Paper Trading Execution if is_paper: cur.execute(""" INSERT INTO trading_orders ( idempotency_key, broker_order_id, provider_code, symbol, exchange, segment, direction, order_type, product_type, price, trigger_price, quantity, filled_quantity, average_price, status, is_paper, stop_loss, target_price, placed_at ) VALUES ( %s, %s, %s, %s, 'NSE', 'EQUITY', %s, 'MARKET', 'INTRADAY', %s, 0.0, %s, %s, %s, 'COMPLETED', 1, %s, %s, NOW() ) """, ( idempotency_key, f"PAPER-{uuid.uuid4().hex[:8].upper()}", "PAPER", symbol, direction, entry_price, quantity, quantity, entry_price, stop_loss, target_1 )) order_id = cur.lastrowid # Create or update active paper position cur.execute(""" INSERT INTO trading_positions ( provider_code, symbol, product_type, quantity, buy_qty, buy_avg_price, current_ltp, unrealized_pnl, stop_loss, target_1, target_2, trailing_stop_loss, status, is_paper, opened_at ) VALUES ( 'PAPER', %s, 'INTRADAY', %s, %s, %s, %s, 0.0, %s, %s, %s, %s, 'OPEN', 1, NOW() ) """, ( symbol, quantity if direction == "BUY" else -quantity, quantity, entry_price, entry_price, stop_loss, target_1, target_2, stop_loss )) conn.close() return { "status": "executed_paper", "order_id": order_id, "idempotency_key": idempotency_key, "symbol": symbol, "entry_price": entry_price, "quantity": quantity, "mode": "PAPER_TRADING" } # Handle Live Broker Order Execution cur.execute("SELECT * FROM broker_credentials WHERE provider_code = %s", (provider_code,)) b_cred = cur.fetchone() if not b_cred or not b_cred.get("encrypted_access_token"): cur.execute(""" INSERT INTO trading_orders ( idempotency_key, broker_order_id, provider_code, symbol, exchange, segment, direction, order_type, product_type, price, trigger_price, quantity, filled_quantity, average_price, status, status_reason, is_paper, stop_loss, target_price, placed_at ) VALUES ( %s, NULL, %s, %s, 'NSE', 'EQUITY', %s, 'MARKET', 'INTRADAY', %s, 0.0, %s, 0, 0.0, 'REJECTED', 'BROKER_NOT_AUTHENTICATED: Configure access token in Broker Management', 0, %s, %s, NOW() ) """, (idempotency_key, provider_code, symbol, direction, entry_price, quantity, stop_loss, target_1)) conn.close() return { "status": "rejected", "reason": f"Broker {provider_code} is not authenticated or access token is expired. Please authenticate via Broker Management." } # Decrypt access token access_token = None try: import base64 from cryptography.hazmat.primitives.ciphers.aead import AESGCM app_key = settings.app_key if hasattr(settings, 'app_key') else 'akdevz_secret_default_key_change_in_production' key = hashlib.sha256(app_key.encode('utf-8')).digest() raw = base64.b64decode(b_cred["encrypted_access_token"]) if len(raw) >= 28: iv = raw[:12] tag = raw[12:28] ct = raw[28:] aesgcm = AESGCM(key) access_token = aesgcm.decrypt(iv, ct + tag, None).decode('utf-8') except Exception as e: logger.warning(f"Failed to decrypt broker access token: {e}") if not access_token: access_token = b_cred.get("encrypted_access_token") # fallback if stored plain # Execute order via selected broker adapter broker_order_id = None broker_err = None try: if provider_code == "FYERS": from python.brokers.fyers_adapter import FyersAdapter app_id = b_cred.get("app_id", "") adapter = FyersAdapter(app_id=app_id, access_token=access_token) fyers_sym = symbol if ":" in symbol else f"NSE:{symbol}-EQ" fyers_payload = { "symbol": fyers_sym, "qty": quantity, "type": 2, # MARKET "side": 1 if direction == "BUY" else -1, "productType": "INTRADAY", "limitPrice": 0, "stopPrice": 0, "validity": "DAY", "disclosedQty": 0, "offlineOrder": False, "stopLoss": stop_loss, "takeProfit": target_1 } res = adapter.place_order(fyers_payload) if res.get("s") == "ok" and res.get("id"): broker_order_id = res["id"] else: broker_err = res.get("message", "Unknown FYERS rejection") elif provider_code == "ANGEL_ONE": from python.brokers.angel_one_adapter import AngelOneAdapter adapter = AngelOneAdapter(client_id=b_cred.get("client_id", ""), jwt_token=access_token) angel_sym = symbol.replace("NSE:", "").replace("-EQ", "") angel_payload = { "variety": "NORMAL", "tradingsymbol": f"{angel_sym}-EQ", "symboltoken": "9999", "transactiontype": direction, "exchange": "NSE", "ordertype": "MARKET", "producttype": "INTRADAY", "duration": "DAY", "price": 0, "quantity": str(quantity) } res = adapter.place_order(angel_payload) if res.get("status") is True and res.get("data", {}).get("orderid"): broker_order_id = res["data"]["orderid"] else: broker_err = res.get("message", "Unknown Angel One rejection") elif provider_code == "UPSTOX": from python.brokers.upstox_adapter import UpstoxAdapter adapter = UpstoxAdapter(api_key=b_cred.get("app_id", ""), access_token=access_token) res = adapter.place_order({ "symbol": symbol, "quantity": quantity, "transaction_type": direction, "order_type": "MARKET", "product": "I" }) if res.get("status") == "success" and res.get("data", {}).get("order_id"): broker_order_id = res["data"]["order_id"] else: broker_err = res.get("message", "Unknown Upstox rejection") elif provider_code == "SHOONYA": from python.brokers.shoonya_adapter import ShoonyaAdapter adapter = ShoonyaAdapter(user_id=b_cred.get("client_id", ""), jkey=access_token) res = adapter.place_order({ "symbol": symbol, "quantity": quantity, "side": direction, "type": "MKT", "product": "I" }) if res.get("stat") == "Ok" and res.get("norenordno"): broker_order_id = res["norenordno"] else: broker_err = res.get("emsg", "Unknown Shoonya rejection") else: broker_err = f"Unsupported live broker code: {provider_code}" except Exception as e: broker_err = str(e) if broker_order_id: cur.execute(""" INSERT INTO trading_orders ( idempotency_key, broker_order_id, provider_code, symbol, exchange, segment, direction, order_type, product_type, price, trigger_price, quantity, filled_quantity, average_price, status, is_paper, stop_loss, target_price, placed_at ) VALUES ( %s, %s, %s, %s, 'NSE', 'EQUITY', %s, 'MARKET', 'INTRADAY', %s, 0.0, %s, %s, %s, 'SUBMITTED', 0, %s, %s, NOW() ) """, ( idempotency_key, broker_order_id, provider_code, symbol, direction, entry_price, quantity, quantity, entry_price, stop_loss, target_1 )) order_id = cur.lastrowid cur.execute(""" INSERT INTO trading_positions ( provider_code, symbol, product_type, quantity, buy_qty, buy_avg_price, current_ltp, unrealized_pnl, stop_loss, target_1, target_2, trailing_stop_loss, status, is_paper, opened_at ) VALUES ( %s, %s, 'INTRADAY', %s, %s, %s, %s, 0.0, %s, %s, %s, %s, 'OPEN', 0, NOW() ) """, ( provider_code, symbol, quantity if direction == "BUY" else -quantity, quantity, entry_price, entry_price, stop_loss, target_1, target_2, stop_loss )) conn.close() return { "status": "executed_live", "order_id": order_id, "broker_order_id": broker_order_id, "symbol": symbol, "provider": provider_code, "quantity": quantity, "mode": "LIVE_BROKER" } else: cur.execute(""" INSERT INTO trading_orders ( idempotency_key, broker_order_id, provider_code, symbol, exchange, segment, direction, order_type, product_type, price, trigger_price, quantity, filled_quantity, average_price, status, status_reason, is_paper, stop_loss, target_price, placed_at ) VALUES ( %s, NULL, %s, %s, 'NSE', 'EQUITY', %s, 'MARKET', 'INTRADAY', %s, 0.0, %s, 0, 0.0, 'REJECTED', %s, 0, %s, %s, NOW() ) """, ( idempotency_key, provider_code, symbol, direction, entry_price, quantity, broker_err or "Broker submission failed", stop_loss, target_1 )) conn.close() return { "status": "rejected", "reason": f"Broker Rejected: {broker_err}" } def monitor_positions_and_trailing_sl(self): """ Continuously monitors open positions: - Updates unrealized P&L against live ticks - Checks Stop Loss and Target hits - Adjusts trailing stop loss when price moves in profit - Automatically closes positions when exit criteria met """ conn = self.get_connection() cur = conn.cursor() cur.execute("SELECT * FROM trading_positions WHERE status = 'OPEN'") positions = cur.fetchall() from python.data.market_feed import MarketFeed for pos in positions: pos_id = pos["id"] sym = pos["symbol"] qty = pos["quantity"] buy_avg = float(pos["buy_avg_price"]) sl = float(pos["stop_loss"] or 0.0) t1 = float(pos["target_1"] or 0.0) trailing_sl = float(pos["trailing_stop_loss"] or sl) quote = MarketFeed.get_real_quote(sym) ltp = float(quote.get("ltp") or 0.0) if ltp <= 0: continue # Update unrealized P&L unrealized = round((ltp - buy_avg) * qty, 2) cur.execute(""" UPDATE trading_positions SET current_ltp = %s, unrealized_pnl = %s, last_synced_at = NOW() WHERE id = %s """, (ltp, unrealized, pos_id)) # Stop-Loss Hit Check if sl > 0 and ltp <= sl: realized = round((sl - buy_avg) * qty, 2) cur.execute(""" UPDATE trading_positions SET status = 'CLOSED', realized_pnl = %s, unrealized_pnl = 0.0, closed_at = NOW() WHERE id = %s """, (realized, pos_id)) logger.info(f"Position {pos_id} ({sym}) STOP LOSS HIT at ₹{sl}. Realized: ₹{realized}") continue # Target 1 Hit Check if t1 > 0 and ltp >= t1: realized = round((t1 - buy_avg) * qty, 2) cur.execute(""" UPDATE trading_positions SET status = 'CLOSED', realized_pnl = %s, unrealized_pnl = 0.0, closed_at = NOW() WHERE id = %s """, (realized, pos_id)) logger.info(f"Position {pos_id} ({sym}) TARGET HIT at ₹{t1}. Realized: ₹{realized}") continue # Trailing Stop-Loss Adjustment (Lock in profits) if ltp > buy_avg + 15.0: new_sl = round(ltp - 15.0, 2) if new_sl > trailing_sl: cur.execute(""" UPDATE trading_positions SET trailing_stop_loss = %s, stop_loss = %s WHERE id = %s """, (new_sl, new_sl, pos_id)) conn.close() def activate_emergency_kill_switch(self) -> Dict[str, Any]: """ EMERGENCY KILL SWITCH: Immediately halts all automated trading, cancels all open pending orders, and records audit alert. """ conn = self.get_connection() cur = conn.cursor() cur.execute("UPDATE risk_configuration SET kill_switch_active = 1, live_execution_enabled = 0 WHERE id = 1") cur.execute("UPDATE trading_orders SET status = 'CANCELLED', status_reason = 'EMERGENCY_KILL_SWITCH' WHERE status IN ('PENDING', 'SUBMITTED', 'OPEN')") cur.execute(""" INSERT INTO audit_logs ( event_type, description, ip_address, created_at ) VALUES ( 'EMERGENCY_KILL_SWITCH_ACTIVATED', 'Trading halted immediately. All open orders cancelled.', '127.0.0.1', NOW() ) """) conn.close() return { "status": "success", "message": "EMERGENCY KILL SWITCH ACTIVATED. Automated execution completely halted." } if __name__ == "__main__": engine = ExecutionEngine() engine.monitor_positions_and_trailing_sl() print(json.dumps({"status": "success", "positions_monitored": True}))