NSE/BSE Algo Trading Bot: Zerodha, Angel One API Integration Guide

Algo Trading System Architecture

# Zerodha Kite Connect Trading Bot
from kiteconnect import KiteConnect, KiteTicker
import pandas as pd
import logging
from datetime import datetime, timedelta

class ZerodhaAlgoBot:
    """NSE/BSE Algo Trading Bot using Zerodha Kite Connect API"""
    
    def __init__(self, api_key: str, api_secret: str, access_token: str):
        self.api_key = api_key
        self.api_secret = api_secret
        self.kite = KiteConnect(api_key=api_key)
        self.kite.set_access_token(access_token)
        self.ticker = None
        self.positions = {}
        
        # Configure logging
        logging.basicConfig(level=logging.INFO)
        self.logger = logging.getLogger(__name__)
        
    def get_instruments(self, exchange: str = "NSE") -> pd.DataFrame:
        """Fetch all instruments for given exchange"""
        instruments = self.kite.instruments(exchange)
        return pd.DataFrame(instruments)
    
    def get_quote(self, symbol: str, exchange: str = "NSE") -> dict:
        """Get current quote for instrument"""
        instrument = f"{exchange}:{symbol}"
        return self.kite.quote([instrument])[instrument]
    
    def get_historical_data(self, instrument_token: int, 
                            from_date: datetime, to_date: datetime,
                            interval: str = "minute") -> pd.DataFrame:
        """Fetch historical OHLCV data"""
        data = self.kite.historical_data(
            instrument_token=instrument_token,
            from_date=from_date,
            to_date=to_date,
            interval=interval
        )
        df = pd.DataFrame(data)
        df.set_index('date', inplace=True)
        return df
    
    def place_order(self, symbol: str, exchange: str, 
                    transaction_type: str, quantity: int,
                    order_type: str = "MARKET", 
                    product: str = "MIS") -> str:
        """Place order on NSE/BSE"""
        try:
            order_id = self.kite.place_order(
                variety=self.kite.VARIETY_REGULAR,
                exchange=exchange,
                tradingsymbol=symbol,
                transaction_type=transaction_type,
                quantity=quantity,
                order_type=order_type,
                product=product
            )
            self.logger.info(f"Order placed: {order_id}")
            return order_id
        except Exception as e:
            self.logger.error(f"Order failed: {e}")
            raise
    
    def place_bracket_order(self, symbol: str, exchange: str,
                            transaction_type: str, quantity: int,
                            price: float, stoploss: float, 
                            target: float) -> str:
        """Place bracket order with SL and target"""
        try:
            order_id = self.kite.place_order(
                variety=self.kite.VARIETY_BO,
                exchange=exchange,
                tradingsymbol=symbol,
                transaction_type=transaction_type,
                quantity=quantity,
                order_type=self.kite.ORDER_TYPE_LIMIT,
                product=self.kite.PRODUCT_MIS,
                price=price,
                stoploss=stoploss,
                squareoff=target
            )
            return order_id
        except Exception as e:
            self.logger.error(f"Bracket order failed: {e}")
            raise
    
    def get_positions(self) -> dict:
        """Get current positions"""
        return self.kite.positions()
    
    def get_holdings(self) -> list:
        """Get portfolio holdings"""
        return self.kite.holdings()

# WebSocket for real-time data
class ZerodhaDataStreamer:
    """Real-time market data streaming via WebSocket"""
    
    def __init__(self, api_key: str, access_token: str):
        self.kws = KiteTicker(api_key, access_token)
        self.ticks_data = {}
        
    def on_ticks(self, ws, ticks):
        """Callback for tick data"""
        for tick in ticks:
            self.ticks_data[tick['instrument_token']] = tick
            self.process_tick(tick)
    
    def on_connect(self, ws, response):
        """Callback on successful connection"""
        print("WebSocket connected")
        
    def start(self, instrument_tokens: list):
        """Start WebSocket streaming"""
        self.kws.on_ticks = self.on_ticks
        self.kws.on_connect = self.on_connect
        self.kws.subscribe(instrument_tokens)
        self.kws.set_mode(self.kws.MODE_FULL, instrument_tokens)
        self.kws.connect()

# Angel One SmartAPI Trading Bot
from SmartApi import SmartConnect
import pyotp
import pandas as pd
import logging

class AngelOneAlgoBot:
    """NSE/BSE Algo Trading Bot using Angel One SmartAPI"""
    
    def __init__(self, api_key: str, client_id: str, 
                 password: str, totp_secret: str):
        self.api_key = api_key
        self.client_id = client_id
        self.password = password
        self.totp_secret = totp_secret
        self.smart_api = SmartConnect(api_key=api_key)
        
        # Initialize session
        self._login()
        
        logging.basicConfig(level=logging.INFO)
        self.logger = logging.getLogger(__name__)
    
    def _login(self):
        """Authenticate with Angel One"""
        totp = pyotp.TOTP(self.totp_secret).now()
        
        session = self.smart_api.generateSession(
            clientCode=self.client_id,
            password=self.password,
            totp=totp
        )
        
        if session['status']:
            self.auth_token = session['data']['jwtToken']
            self.refresh_token = session['data']['refreshToken']
            self.feed_token = self.smart_api.getfeedToken()
            print("Login successful")
        else:
            raise Exception(f"Login failed: {session}")
    
    def get_ltp(self, exchange: str, symbol: str, 
                token: str) -> float:
        """Get last traded price"""
        ltp_data = self.smart_api.ltpData(exchange, symbol, token)
        return ltp_data['data']['ltp']
    
    def place_order(self, symbol: str, token: str,
                    transaction_type: str, exchange: str,
                    quantity: int, order_type: str = "MARKET",
                    product_type: str = "INTRADAY",
                    price: float = 0) -> dict:
        """Place order on NSE/BSE via Angel One"""
        
        order_params = {
            "variety": "NORMAL",
            "tradingsymbol": symbol,
            "symboltoken": token,
            "transactiontype": transaction_type,
            "exchange": exchange,
            "ordertype": order_type,
            "producttype": product_type,
            "duration": "DAY",
            "quantity": str(quantity)
        }
        
        if order_type == "LIMIT":
            order_params["price"] = str(price)
        
        try:
            response = self.smart_api.placeOrder(order_params)
            self.logger.info(f"Order placed: {response}")
            return response
        except Exception as e:
            self.logger.error(f"Order failed: {e}")
            raise
    
    def get_historical_data(self, exchange: str, 
                            symbol: str, token: str,
                            interval: str, from_date: str,
                            to_date: str) -> pd.DataFrame:
        """Fetch historical candle data"""
        
        params = {
            "exchange": exchange,
            "symboltoken": token,
            "interval": interval,
            "fromdate": from_date,
            "todate": to_date
        }
        
        response = self.smart_api.getCandleData(params)
        
        if response['status']:
            df = pd.DataFrame(
                response['data'],
                columns=['timestamp', 'open', 'high', 
                         'low', 'close', 'volume']
            )
            df['timestamp'] = pd.to_datetime(df['timestamp'])
            return df
        return pd.DataFrame()
    
    def get_positions(self) -> dict:
        """Get current positions"""
        return self.smart_api.position()
    
    def get_holdings(self) -> dict:
        """Get portfolio holdings"""
        return self.smart_api.holding()

# NIFTY Straddle/Strangle Strategy
class NiftyOptionsStrategy:
    """Options selling strategy for NIFTY index"""
    
    def __init__(self, broker_api):
        self.api = broker_api
        self.lot_size = 50  # NIFTY lot size
        self.max_loss_per_trade = 5000
        
    def get_atm_strike(self, spot_price: float) -> int:
        """Calculate ATM strike price"""
        return round(spot_price / 50) * 50
    
    def calculate_strangle_strikes(self, spot_price: float, 
                                    distance: int = 200) -> tuple:
        """Get OTM call and put strikes for strangle"""
        atm = self.get_atm_strike(spot_price)
        call_strike = atm + distance
        put_strike = atm - distance
        return call_strike, put_strike
    
    def execute_short_strangle(self, expiry: str, 
                               lots: int = 1) -> dict:
        """Execute short strangle strategy"""
        
        # Get NIFTY spot price
        nifty_ltp = self.api.get_quote("NIFTY 50", "NSE")['last_price']
        
        # Calculate strikes
        call_strike, put_strike = self.calculate_strangle_strikes(nifty_ltp)
        
        # Generate option symbols
        call_symbol = f"NIFTY{expiry}{call_strike}CE"
        put_symbol = f"NIFTY{expiry}{put_strike}PE"
        
        quantity = lots * self.lot_size
        
        # Sell call option
        call_order = self.api.place_order(
            symbol=call_symbol,
            exchange="NFO",
            transaction_type="SELL",
            quantity=quantity,
            order_type="MARKET",
            product="NRML"
        )
        
        # Sell put option
        put_order = self.api.place_order(
            symbol=put_symbol,
            exchange="NFO",
            transaction_type="SELL",
            quantity=quantity,
            order_type="MARKET",
            product="NRML"
        )
        
        return {
            'call_order': call_order,
            'put_order': put_order,
            'call_strike': call_strike,
            'put_strike': put_strike
        }
    
    def calculate_max_profit(self, call_premium: float, 
                             put_premium: float, lots: int) -> float:
        """Calculate maximum profit for short strangle"""
        total_premium = (call_premium + put_premium) * lots * self.lot_size
        return total_premium

class IndianMarketRiskManager:
    """Risk management for NSE/BSE trading"""
    
    def __init__(self, capital: float, max_daily_loss_pct: float = 5.0):
        self.capital = capital
        self.max_daily_loss = capital * (max_daily_loss_pct / 100)
        self.daily_pnl = 0
        self.trading_enabled = True
        
    def calculate_position_size(self, risk_per_trade_pct: float,
                                 stop_loss_points: float,
                                 lot_size: int) -> int:
        """Calculate number of lots based on risk"""
        risk_amount = self.capital * (risk_per_trade_pct / 100)
        risk_per_lot = stop_loss_points * lot_size
        lots = int(risk_amount / risk_per_lot)
        return max(1, lots)
    
    def check_circuit_breaker(self, symbol_data: dict) -> bool:
        """Check if stock is in circuit limit"""
        ltp = symbol_data['last_price']
        upper_circuit = symbol_data.get('upper_circuit_limit')
        lower_circuit = symbol_data.get('lower_circuit_limit')
        
        if upper_circuit and ltp >= upper_circuit:
            return True  # Upper circuit hit
        if lower_circuit and ltp <= lower_circuit:
            return True  # Lower circuit hit
        return False
    
    def update_pnl(self, pnl: float):
        """Update daily P&L and check limits"""
        self.daily_pnl += pnl
        
        if self.daily_pnl <= -self.max_daily_loss:
            self.trading_enabled = False
            print("Daily loss limit reached. Trading disabled.")
    
    def is_market_hours(self) -> bool:
        """Check if within NSE/BSE trading hours"""
        from datetime import datetime, time
        
        now = datetime.now().time()
        market_open = time(9, 15)
        market_close = time(15, 30)
        
        return market_open <= now <= market_close
    
    def can_trade(self) -> bool:
        """Check if trading is allowed"""
        return self.trading_enabled and self.is_market_hours()