Automating Trades with Python and Broker APIs: The Complete Indian Guide
Master the art of automated trading by connecting your Python strategies to Zerodha, Upstox, and other Indian brokers. Complete with authentication, order management, and risk controls.
The jump from backtested strategies to live automated trading is where most quantitative traders stumble. In India’s unique market structure, with its specific broker APIs, authentication requirements, and regulatory constraints, automation requires careful planning and robust implementation.
This comprehensive guide walks you through the entire process of automating your trading strategies using Python and Indian broker APIs, from authentication to order execution to risk management.
Why Automate Trading in India?
The Case for Automation
Speed: Execute orders in milliseconds instead of minutes Discipline: Eliminate emotional decision-making Scalability: Monitor hundreds of instruments simultaneously Consistency: Execute strategies exactly as designed Cost efficiency: Reduce manual errors and missed opportunities
The Indian Context
Indian markets present unique challenges:
- Limited API access (not all brokers offer programmatic trading)
- Two-factor authentication requirements
- Margin calculations differ across brokers
- SEBI regulations on algo trading
- Market hours and circuit breakers
Part 1: Choosing Your Broker API
Popular Options in India
1. Zerodha Kite Connect ⭐ Most Popular
- Pros: Excellent documentation, stable API, WebSocket support, large community
- Cons: ₹2000/month for API access, rate limits
- Best for: Serious retail traders and small funds
2. Upstox API
- Pros: Free API access, modern REST interface, good documentation
- Cons: Smaller community, occasional stability issues
- Best for: Cost-conscious traders, beginners
3. Angel One SmartAPI
- Pros: Free API, historical data access, decent documentation
- Cons: Limited WebSocket capabilities
- Best for: Backtesting and strategy research
4. Fyers API
- Pros: Clean REST API, competitive pricing
- Cons: Smaller user base
- Best for: Traders wanting alternatives to Zerodha
5. Interactive Brokers (IBKR)
- Pros: Global market access, professional-grade infrastructure
- Cons: Complex setup, higher costs, not optimized for Indian retail
- Best for: Professional traders, international diversification
Decision Matrix
| Feature | Zerodha | Upstox | Angel One | Fyers |
|---|---|---|---|---|
| API Cost | ₹2000/mo | Free | Free | ₹1000/mo |
| WebSocket | ✅ | ✅ | Limited | ✅ |
| Documentation | Excellent | Good | Good | Good |
| Community | Large | Medium | Medium | Small |
| Stability | High | Medium | Medium | Medium |
Part 2: Authentication and Connection
Zerodha Kite Connect Setup
Step 1: Register for API Access
# 1. Sign up at https://developers.kite.trade/
# 2. Create an app to get API key and secret
# 3. Set redirect URL (can be http://localhost for testing)
Step 2: Implement OAuth Flow
from kiteconnect import KiteConnect
import logging
# Configuration
API_KEY = "your_api_key"
API_SECRET = "your_api_secret"
# Initialize KiteConnect
kite = KiteConnect(api_key=API_KEY)
def authenticate():
"""
Manual authentication flow
Run this once per day to get access token
"""
# Generate login URL
login_url = kite.login_url()
print(f"Open this URL in browser: {login_url}")
# After login, you'll be redirected to your redirect_url with request_token
request_token = input("Enter request_token from URL: ")
# Generate access token
data = kite.generate_session(request_token, api_secret=API_SECRET)
access_token = data["access_token"]
# Save access token (valid till 6 AM next day)
with open('access_token.txt', 'w') as f:
f.write(access_token)
return access_token
# Set access token
try:
with open('access_token.txt', 'r') as f:
access_token = f.read().strip()
kite.set_access_token(access_token)
# Verify connection
profile = kite.profile()
print(f"Connected as: {profile['user_name']}")
except FileNotFoundError:
access_token = authenticate()
kite.set_access_token(access_token)
Step 3: Automated Token Management
import pyotp
from selenium import webdriver
from selenium.webdriver.common.by import By
import time
class ZerodhaAutoLogin:
"""
Automated login using TOTP (Time-based OTP)
WARNING: Store credentials securely!
"""
def __init__(self, user_id, password, totp_secret, api_key, api_secret):
self.user_id = user_id
self.password = password
self.totp_secret = totp_secret
self.api_key = api_key
self.api_secret = api_secret
self.kite = KiteConnect(api_key=api_key)
def get_access_token(self):
"""
Fully automated login and token generation
"""
# Setup Chrome in headless mode
options = webdriver.ChromeOptions()
options.add_argument('--headless')
driver = webdriver.Chrome(options=options)
try:
# Navigate to login URL
login_url = self.kite.login_url()
driver.get(login_url)
time.sleep(2)
# Enter user ID
driver.find_element(By.ID, "userid").send_keys(self.user_id)
driver.find_element(By.ID, "password").send_keys(self.password)
driver.find_element(By.XPATH, "//button[@type='submit']").click()
time.sleep(2)
# Enter TOTP
totp = pyotp.TOTP(self.totp_secret)
current_otp = totp.now()
driver.find_element(By.ID, "totp").send_keys(current_otp)
driver.find_element(By.XPATH, "//button[@type='submit']").click()
time.sleep(3)
# Extract request token from redirect URL
current_url = driver.current_url
request_token = current_url.split('request_token=')[1].split('&')[0]
# Generate access token
data = self.kite.generate_session(request_token, api_secret=self.api_secret)
access_token = data["access_token"]
return access_token
finally:
driver.quit()
def login(self):
"""
Main login method with error handling
"""
try:
# Try reading existing token
with open('zerodha_token.txt', 'r') as f:
token_data = f.read().strip().split(',')
access_token = token_data[0]
timestamp = float(token_data[1])
# Check if token is still valid (expires at 6 AM IST)
from datetime import datetime
import pytz
ist = pytz.timezone('Asia/Kolkata')
current_time = datetime.now(ist)
token_time = datetime.fromtimestamp(timestamp, ist)
# If token is from today and current time < 6 AM, use existing token
if current_time.date() == token_time.date() and current_time.hour < 6:
self.kite.set_access_token(access_token)
logging.info("Using existing access token")
return self.kite
except (FileNotFoundError, IndexError, ValueError):
pass
# Get new token
logging.info("Generating new access token")
access_token = self.get_access_token()
self.kite.set_access_token(access_token)
# Save token with timestamp
timestamp = time.time()
with open('zerodha_token.txt', 'w') as f:
f.write(f"{access_token},{timestamp}")
return self.kite
# Usage
auto_login = ZerodhaAutoLogin(
user_id="YOUR_USER_ID",
password="YOUR_PASSWORD",
totp_secret="YOUR_TOTP_SECRET", # Get from Kite TOTP setup
api_key="YOUR_API_KEY",
api_secret="YOUR_API_SECRET"
)
kite = auto_login.login()
Upstox API Authentication
from upstox_client import Configuration, LoginApi, ApiClient
import requests
API_KEY = "your_api_key"
API_SECRET = "your_api_secret"
REDIRECT_URI = "http://localhost"
def authenticate_upstox():
"""
Upstox OAuth2 flow
"""
# Step 1: Get authorization URL
auth_url = f"https://api.upstox.com/v2/login/authorization/dialog?response_type=code&client_id={API_KEY}&redirect_uri={REDIRECT_URI}"
print(f"Open this URL: {auth_url}")
redirect_url = input("Enter the redirect URL after authorization: ")
# Extract code
auth_code = redirect_url.split('code=')[1].split('&')[0]
# Step 2: Exchange code for access token
token_url = "https://api.upstox.com/v2/login/authorization/token"
payload = {
'code': auth_code,
'client_id': API_KEY,
'client_secret': API_SECRET,
'redirect_uri': REDIRECT_URI,
'grant_type': 'authorization_code'
}
response = requests.post(token_url, data=payload)
access_token = response.json()['access_token']
# Save token
with open('upstox_token.txt', 'w') as f:
f.write(access_token)
return access_token
# Initialize API client
access_token = authenticate_upstox()
configuration = Configuration()
configuration.access_token = access_token
api_client = ApiClient(configuration)
Part 3: Order Execution
Basic Order Types
class OrderManager:
"""
Unified order management across different order types
"""
def __init__(self, kite):
self.kite = kite
self.orders = {}
def place_market_order(self, symbol, quantity, transaction_type, product="MIS"):
"""
Market order - executes at best available price
Use for: Urgent executions, high liquidity stocks
"""
try:
order_id = self.kite.place_order(
variety=self.kite.VARIETY_REGULAR,
exchange=self.kite.EXCHANGE_NSE,
tradingsymbol=symbol,
transaction_type=transaction_type, # BUY or SELL
quantity=quantity,
product=product, # MIS (intraday) or CNC (delivery)
order_type=self.kite.ORDER_TYPE_MARKET
)
logging.info(f"Market order placed: {order_id}")
self.orders[order_id] = {
'symbol': symbol,
'quantity': quantity,
'type': transaction_type,
'product': product
}
return order_id
except Exception as e:
logging.error(f"Order placement failed: {e}")
return None
def place_limit_order(self, symbol, quantity, price, transaction_type, product="MIS"):
"""
Limit order - executes only at specified price or better
Use for: Better price execution, non-urgent trades
"""
try:
order_id = self.kite.place_order(
variety=self.kite.VARIETY_REGULAR,
exchange=self.kite.EXCHANGE_NSE,
tradingsymbol=symbol,
transaction_type=transaction_type,
quantity=quantity,
product=product,
order_type=self.kite.ORDER_TYPE_LIMIT,
price=price
)
logging.info(f"Limit order placed: {order_id} at ₹{price}")
return order_id
except Exception as e:
logging.error(f"Limit order failed: {e}")
return None
def place_bracket_order(self, symbol, quantity, price, stoploss, target, transaction_type="BUY"):
"""
Bracket order - entry + SL + target in one order
Use for: Intraday trading with predefined risk/reward
Note: Not all brokers support bracket orders
Check broker documentation
"""
try:
order_id = self.kite.place_order(
variety=self.kite.VARIETY_BO,
exchange=self.kite.EXCHANGE_NSE,
tradingsymbol=symbol,
transaction_type=transaction_type,
quantity=quantity,
product=self.kite.PRODUCT_MIS,
order_type=self.kite.ORDER_TYPE_LIMIT,
price=price,
stoploss=stoploss, # Points away from entry
squareoff=target, # Points away from entry
trailing_stoploss=2 # Optional trailing SL
)
logging.info(f"Bracket order placed: {order_id}")
return order_id
except Exception as e:
logging.error(f"Bracket order failed: {e}")
return None
def modify_order(self, order_id, quantity=None, price=None):
"""
Modify pending order
"""
try:
self.kite.modify_order(
variety=self.kite.VARIETY_REGULAR,
order_id=order_id,
quantity=quantity,
price=price,
order_type=self.kite.ORDER_TYPE_LIMIT if price else self.kite.ORDER_TYPE_MARKET
)
logging.info(f"Order modified: {order_id}")
except Exception as e:
logging.error(f"Order modification failed: {e}")
def cancel_order(self, order_id):
"""
Cancel pending order
"""
try:
self.kite.cancel_order(
variety=self.kite.VARIETY_REGULAR,
order_id=order_id
)
logging.info(f"Order cancelled: {order_id}")
except Exception as e:
logging.error(f"Order cancellation failed: {e}")
def get_order_status(self, order_id):
"""
Check order execution status
"""
try:
orders = self.kite.orders()
for order in orders:
if order['order_id'] == order_id:
return order['status'] # COMPLETE, OPEN, CANCELLED, REJECTED
return None
except Exception as e:
logging.error(f"Failed to get order status: {e}")
return None
# Usage
order_manager = OrderManager(kite)
# Place market order
order_id = order_manager.place_market_order(
symbol="RELIANCE",
quantity=10,
transaction_type=kite.TRANSACTION_TYPE_BUY,
product=kite.PRODUCT_CNC
)
# Place limit order
limit_order = order_manager.place_limit_order(
symbol="TCS",
quantity=5,
price=3650.00,
transaction_type=kite.TRANSACTION_TYPE_BUY
)
# Check status
status = order_manager.get_order_status(order_id)
print(f"Order status: {status}")
Smart Order Execution
class SmartOrderExecutor:
"""
Intelligent order execution with slippage control
"""
def __init__(self, kite, max_slippage_pct=0.5):
self.kite = kite
self.max_slippage_pct = max_slippage_pct
def execute_with_slippage_control(self, symbol, quantity, transaction_type):
"""
Place limit order near market price to control slippage
"""
# Get current market price
quote = self.kite.quote(f"NSE:{symbol}")[f"NSE:{symbol}"]
ltp = quote['last_price']
# Calculate limit price based on transaction type
if transaction_type == self.kite.TRANSACTION_TYPE_BUY:
# Buy slightly above LTP to ensure execution
limit_price = ltp * (1 + self.max_slippage_pct / 100)
else:
# Sell slightly below LTP
limit_price = ltp * (1 - self.max_slippage_pct / 100)
# Round to tick size (0.05 for most stocks)
limit_price = round(limit_price / 0.05) * 0.05
# Place limit order
order_id = self.kite.place_order(
variety=self.kite.VARIETY_REGULAR,
exchange=self.kite.EXCHANGE_NSE,
tradingsymbol=symbol,
transaction_type=transaction_type,
quantity=quantity,
product=self.kite.PRODUCT_MIS,
order_type=self.kite.ORDER_TYPE_LIMIT,
price=limit_price
)
logging.info(f"Order placed: {symbol} @ ₹{limit_price} (LTP: ₹{ltp})")
# Monitor for 30 seconds
import time
for _ in range(6): # Check every 5 seconds
time.sleep(5)
status = self.get_order_status(order_id)
if status == "COMPLETE":
logging.info(f"Order executed: {order_id}")
return order_id
elif status == "REJECTED":
logging.error(f"Order rejected: {order_id}")
return None
# If not executed, modify to market order
logging.warning(f"Order {order_id} not filled, converting to market order")
self.kite.modify_order(
variety=self.kite.VARIETY_REGULAR,
order_id=order_id,
order_type=self.kite.ORDER_TYPE_MARKET
)
return order_id
Part 4: Real-Time Data Streaming
WebSocket Implementation
from kiteconnect import KiteTicker
import logging
class MarketDataHandler:
"""
Real-time market data streaming with WebSocket
"""
def __init__(self, api_key, access_token):
self.kws = KiteTicker(api_key, access_token)
self.subscribed_tokens = []
self.tick_data = {}
# Set callbacks
self.kws.on_ticks = self.on_ticks
self.kws.on_connect = self.on_connect
self.kws.on_close = self.on_close
self.kws.on_error = self.on_error
def on_ticks(self, ws, ticks):
"""
Process incoming tick data
"""
for tick in ticks:
instrument_token = tick['instrument_token']
self.tick_data[instrument_token] = {
'ltp': tick['last_price'],
'volume': tick['volume'],
'bid': tick.get('depth', {}).get('buy', [{}])[0].get('price'),
'ask': tick.get('depth', {}).get('sell', [{}])[0].get('price'),
'timestamp': tick['timestamp']
}
# Call strategy logic
self.process_tick(tick)
def on_connect(self, ws, response):
"""
Subscribe to instruments on connection
"""
logging.info("WebSocket connected")
if self.subscribed_tokens:
ws.subscribe(self.subscribed_tokens)
ws.set_mode(ws.MODE_FULL, self.subscribed_tokens)
def on_close(self, ws, code, reason):
"""
Handle WebSocket disconnection
"""
logging.warning(f"WebSocket closed: {code} - {reason}")
# Implement reconnection logic
def on_error(self, ws, code, reason):
"""
Handle WebSocket errors
"""
logging.error(f"WebSocket error: {code} - {reason}")
def subscribe(self, instrument_tokens):
"""
Subscribe to instruments
"""
self.subscribed_tokens.extend(instrument_tokens)
def start(self):
"""
Start WebSocket connection
"""
self.kws.connect(threaded=True)
def stop(self):
"""
Stop WebSocket connection
"""
self.kws.close()
def process_tick(self, tick):
"""
Override this method in subclass to implement strategy logic
"""
pass
# Usage
class TradingStrategy(MarketDataHandler):
"""
Custom trading strategy
"""
def __init__(self, api_key, access_token, kite):
super().__init__(api_key, access_token)
self.kite = kite
self.positions = {}
def process_tick(self, tick):
"""
Strategy logic on each tick
"""
symbol = tick['tradable']
ltp = tick['last_price']
# Example: Simple moving average crossover
# (In reality, you'd maintain price history)
if self.should_buy(tick):
self.place_order(symbol, 'BUY')
elif self.should_sell(tick):
self.place_order(symbol, 'SELL')
# Initialize
strategy = TradingStrategy(API_KEY, access_token, kite)
strategy.subscribe([738561, 341249]) # Reliance, TCS
strategy.start()
Part 5: Risk Management
Position Management
class RiskManager:
"""
Comprehensive risk management system
"""
def __init__(self, kite, max_position_size=100000, max_loss_per_trade=2000):
self.kite = kite
self.max_position_size = max_position_size
self.max_loss_per_trade = max_loss_per_trade
self.daily_pnl = 0
self.max_daily_loss = -10000
def check_margins(self):
"""
Check available margin before placing order
"""
try:
margins = self.kite.margins()
available_cash = margins['equity']['available']['cash']
return available_cash
except Exception as e:
logging.error(f"Failed to fetch margins: {e}")
return 0
def calculate_position_size(self, symbol, entry_price, stop_loss_price):
"""
Calculate position size based on risk per trade
"""
# Risk per share
risk_per_share = abs(entry_price - stop_loss_price)
# Maximum shares based on risk
max_shares = int(self.max_loss_per_trade / risk_per_share)
# Maximum shares based on position size limit
max_shares_by_capital = int(self.max_position_size / entry_price)
# Take minimum
position_size = min(max_shares, max_shares_by_capital)
logging.info(f"Calculated position size for {symbol}: {position_size} shares")
return position_size
def check_daily_loss_limit(self):
"""
Check if daily loss limit exceeded
"""
self.update_daily_pnl()
if self.daily_pnl <= self.max_daily_loss:
logging.warning(f"Daily loss limit reached: ₹{self.daily_pnl}")
self.close_all_positions()
return False
return True
def update_daily_pnl(self):
"""
Update daily P&L from positions
"""
try:
positions = self.kite.positions()['net']
self.daily_pnl = sum([pos['pnl'] for pos in positions])
except Exception as e:
logging.error(f"Failed to update P&L: {e}")
def close_all_positions(self):
"""
Emergency position exit
"""
try:
positions = self.kite.positions()['net']
for pos in positions:
if pos['quantity'] != 0:
# Determine transaction type
transaction_type = self.kite.TRANSACTION_TYPE_SELL if pos['quantity'] > 0 else self.kite.TRANSACTION_TYPE_BUY
# Place market order to close
self.kite.place_order(
variety=self.kite.VARIETY_REGULAR,
exchange=pos['exchange'],
tradingsymbol=pos['tradingsymbol'],
transaction_type=transaction_type,
quantity=abs(pos['quantity']),
product=pos['product'],
order_type=self.kite.ORDER_TYPE_MARKET
)
logging.info(f"Closed position: {pos['tradingsymbol']}")
except Exception as e:
logging.error(f"Failed to close positions: {e}")
# Usage
risk_manager = RiskManager(kite)
# Before placing trade
available_margin = risk_manager.check_margins()
if available_margin > 50000:
position_size = risk_manager.calculate_position_size("RELIANCE", 2450, 2400)
# Place order with calculated size
Part 6: Production-Ready Implementation
Complete Trading Bot
import logging
from datetime import datetime
import pytz
import time
class AutomatedTradingBot:
"""
Production-ready automated trading system
"""
def __init__(self, config):
self.config = config
self.kite = self.initialize_connection()
self.risk_manager = RiskManager(self.kite)
self.order_manager = OrderManager(self.kite)
self.is_running = False
# Setup logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('trading_bot.log'),
logging.StreamHandler()
]
)
def initialize_connection(self):
"""
Initialize broker connection
"""
# ... (authentication logic from earlier)
pass
def is_market_open(self):
"""
Check if market is open for trading
"""
ist = pytz.timezone('Asia/Kolkata')
now = datetime.now(ist)
# Check if weekend
if now.weekday() >= 5: # Saturday or Sunday
return False
# Check trading hours (9:15 AM to 3:30 PM IST)
market_open = now.replace(hour=9, minute=15, second=0)
market_close = now.replace(hour=15, minute=30, second=0)
return market_open <= now <= market_close
def run_strategy(self):
"""
Main strategy execution loop
"""
# Check risk limits
if not self.risk_manager.check_daily_loss_limit():
logging.warning("Trading halted due to daily loss limit")
return
# Fetch market data
# ... (your strategy logic)
# Generate signals
signals = self.generate_signals()
# Execute trades
for signal in signals:
self.execute_trade(signal)
def execute_trade(self, signal):
"""
Execute trade based on signal
"""
symbol = signal['symbol']
action = signal['action'] # BUY or SELL
quantity = signal['quantity']
# Pre-trade checks
available_margin = self.risk_manager.check_margins()
if available_margin < quantity * signal['price']:
logging.warning(f"Insufficient margin for {symbol}")
return
# Place order
order_id = self.order_manager.place_market_order(
symbol=symbol,
quantity=quantity,
transaction_type=action
)
if order_id:
logging.info(f"Trade executed: {action} {quantity} {symbol}")
def start(self):
"""
Start the trading bot
"""
logging.info("Starting automated trading bot")
self.is_running = True
while self.is_running:
try:
if self.is_market_open():
self.run_strategy()
time.sleep(60) # Run every minute
else:
logging.info("Market closed, waiting...")
time.sleep(300) # Check every 5 minutes
except Exception as e:
logging.error(f"Error in main loop: {e}")
time.sleep(60)
def stop(self):
"""
Stop the trading bot
"""
logging.info("Stopping trading bot")
self.is_running = False
# Close all positions if configured
if self.config.get('close_positions_on_stop'):
self.risk_manager.close_all_positions()
# Configuration
config = {
'api_key': 'your_api_key',
'api_secret': 'your_api_secret',
'user_id': 'your_user_id',
'password': 'your_password',
'totp_secret': 'your_totp_secret',
'max_position_size': 100000,
'max_loss_per_trade': 2000,
'close_positions_on_stop': True
}
# Start bot
bot = AutomatedTradingBot(config)
bot.start()
Conclusion
Automating trades with Python and Indian broker APIs transforms how you interact with markets. Key takeaways:
- Choose the right broker API based on your needs and budget
- Implement robust authentication with token management
- Use smart order execution to control slippage and costs
- Build comprehensive risk management to protect capital
- Test extensively before going live with real money
Ready to automate your trading? Contact us for professional automated trading solutions tailored to Indian markets.