"""Zerodha broker implementation""" import time from typing import Dict, List, Optional from kiteconnect import KiteConnect from .base_broker import BaseBroker from utils.logger import TradingLogger class ZerodhaBroker(BaseBroker): """Zerodha specific implementation""" def __init__(self, credentials: Dict, client_code: str): super().__init__(credentials, client_code) self.api_key = credentials.get('api_key') self.access_token = credentials.get('access_token') self.kite = None self.logger = TradingLogger(client_code) def login(self) -> bool: """Login to Zerodha""" try: self.kite = KiteConnect(api_key=self.api_key) self.kite.set_access_token(self.access_token) profile = self.kite.profile() self.is_logged_in = True self.logger.success(f"Zerodha login successful for user: {profile.get('user_name')}") return True except Exception as e: self.logger.error(f"Zerodha login failed: {e}") return False def place_order(self, symbol: str, action: str, quantity: int, exchange: str, expiry: str, remarks: str) -> Optional[str]: """Place order on Zerodha""" try: # Construct trading symbol if symbol == 'SENSEX': trading_symbol = symbol else: instrument_post = symbol.replace("NIFTY", "") trading_symbol = f"NIFTY{expiry}{instrument_post}" transaction_type = (KiteConnect.TRANSACTION_TYPE_BUY if action.upper() == 'BUY' else KiteConnect.TRANSACTION_TYPE_SELL) order_id = self.kite.place_order( variety=KiteConnect.VARIETY_REGULAR, exchange=exchange, tradingsymbol=trading_symbol, transaction_type=transaction_type, quantity=quantity, product=KiteConnect.PRODUCT_NRML, order_type=KiteConnect.ORDER_TYPE_MARKET, price="0.0", validity=KiteConnect.VALIDITY_DAY, market_protection="10", tag=remarks[-20:] if remarks else "TAG" ) self.logger.success(f"Zerodha order placed: ID={order_id}, Symbol={trading_symbol}, Qty={quantity}") return order_id except Exception as e: self.logger.error(f"Zerodha order failed: {e}") return None def place_profit_order(self, trading_symbol: str, quantity: int, exchange: str, transaction_type: str, remarks: str) -> Optional[str]: """Place profit booking order on Zerodha""" try: order_id = self.kite.place_order( variety=KiteConnect.VARIETY_REGULAR, exchange=exchange, tradingsymbol=trading_symbol, transaction_type=transaction_type, quantity=quantity, product=KiteConnect.PRODUCT_NRML, order_type=KiteConnect.ORDER_TYPE_MARKET, price="0.0", validity=KiteConnect.VALIDITY_DAY, market_protection="10", tag=remarks ) self.logger.success(f"Zerodha profit order placed: ID={order_id}, Symbol={trading_symbol}") return order_id except Exception as e: self.logger.error(f"Zerodha profit order failed: {e}") return None def get_positions(self) -> List[Dict]: """Get current positions from Zerodha""" try: positions_data = self.kite.positions() return positions_data.get('net', []) except Exception as e: self.logger.error(f"Failed to get positions: {e}") return [] def get_orders(self) -> List[Dict]: """Get order book from Zerodha""" try: return self.kite.orders() except Exception as e: self.logger.error(f"Failed to get orders: {e}") return [] def get_quote(self, exchange: str, trading_symbol: str) -> Dict: """Get market quote from Zerodha""" try: instruments = self.kite.instruments(exchange) instrument_token = None for inst in instruments: if inst['tradingsymbol'] == trading_symbol: instrument_token = inst['instrument_token'] break if instrument_token: quote = self.kite.quote([instrument_token]) return { 'last_price': float(quote[str(instrument_token)]['last_price']), 'change': float(quote[str(instrument_token)].get('change', 0)) } return {'last_price': 0, 'change': 0} except Exception as e: self.logger.error(f"Failed to get quote for {trading_symbol}: {e}") return {'last_price': 0, 'change': 0} def get_realized_pnl(self) -> float: """Get total realized PnL from Zerodha""" try: positions = self.kite.positions() net_positions = positions.get('net', []) total_pnl = sum(item.get('pnl', 0) for item in net_positions) return total_pnl except Exception as e: self.logger.error(f"Failed to get realized PnL: {e}") return 0.0 def close_all_positions(self, max_profit_target: int, max_loss_limit: int, dynamic_profit_enabled: bool) -> bool: """Close all open positions based on limits""" if max_profit_target == 0 and max_loss_limit == 0: return False try: positions = self.get_positions() if all(pos.get('quantity', 0) == 0 for pos in positions): total_pnl = self.get_realized_pnl() if total_pnl > max_profit_target: self.logger.info(f"MAX PROFIT reached: {total_pnl} > {max_profit_target}") return True elif total_pnl < -max_loss_limit: self.logger.info(f"MAX LOSS reached: {total_pnl} < -{max_loss_limit}") return True return False if not dynamic_profit_enabled: return False total_pnl = sum(pos.get('pnl', 0) for pos in positions) should_close = False marker = "" if total_pnl > max_profit_target: should_close = True marker = "ProfitLimit" elif total_pnl < -max_loss_limit: should_close = True marker = "LossLimit" if should_close: self.logger.info(f"FORCE CLOSE ALL POSITIONS - {marker}") for position in positions: quantity = position.get('quantity', 0) if quantity != 0: trading_symbol = position.get('tradingsymbol') exchange = position.get('exchange') transaction_type = KiteConnect.TRANSACTION_TYPE_SELL self.place_profit_order( trading_symbol=trading_symbol, quantity=quantity, exchange=exchange, transaction_type=transaction_type, remarks=f"FORCE_CLOSE_{marker}" ) time.sleep(0.5) return True return False except Exception as e: self.logger.error(f"Failed to close positions: {e}") return False