Complete source code
# Complete NSE momentum strategy with backtesting and live trading
# Broker: Zerodha via kiteconnect
# Strategy: Buy top 5 momentum stocks from Nifty 50 every week
import yfinance as yf
import pandas as pd
import numpy as np
from kiteconnect import KiteConnect # pip install kiteconnect
from datetime import datetime, timedelta
import logging
import time
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# ── CONFIG ───────────────────────────────────────────────────────────
NIFTY50_STOCKS = [
"RELIANCE", "TCS", "HDFCBANK", "ICICIBANK", "INFY",
"HINDUNILVR", "ITC", "SBIN", "BAJFINANCE", "BHARTIARTL",
"KOTAKBANK", "WIPRO", "LT", "ASIANPAINT", "AXISBANK",
"HCLTECH", "SUNPHARMA", "TATAMOTORS", "NESTLEIND", "TITAN"
]
MAX_POSITIONS = 5 # Hold top 5 stocks
CAPITAL_PER_STOCK = 10000 # ₹10,000 per stock
STOP_LOSS_PCT = 0.05 # 5% stop loss
PROFIT_TARGET_PCT = 0.12 # 12% profit target
LOOKBACK_DAYS = 20 # 20-day momentum
# ── MOMENTUM CALCULATION ─────────────────────────────────────────────
def calculate_momentum(symbol: str, days: int = 20) -> float:
"""Calculate momentum score: price change % + volume factor"""
try:
ticker = yf.Ticker(f"{symbol}.NS")
df = ticker.history(period="3mo", interval="1d")
if len(df) < days:
return 0
# Price momentum
price_momentum = (df['Close'].iloc[-1] / df['Close'].iloc[-days] - 1) * 100
# Volume momentum (recent volume vs average)
recent_volume = df['Volume'].iloc[-5:].mean()
avg_volume = df['Volume'].iloc[-days:].mean()
volume_factor = recent_volume / avg_volume if avg_volume > 0 else 1
# Combined score (price momentum weighted by volume)
momentum_score = price_momentum * volume_factor
return momentum_score
except Exception as e:
logger.error(f"Error calculating momentum for {symbol}: {e}")
return 0
def get_top_momentum_stocks(n: int = 5) -> list:
"""Rank all Nifty 50 stocks by momentum and return top N"""
scores = {}
for symbol in NIFTY50_STOCKS:
score = calculate_momentum(symbol, LOOKBACK_DAYS)
scores[symbol] = score
time.sleep(0.2) # Rate limiting
# Sort by score descending
sorted_stocks = sorted(scores.items(), key=lambda x: x[1], reverse=True)
logger.info("Momentum Rankings:")
for symbol, score in sorted_stocks[:10]:
logger.info(f" {symbol}: {score:.2f}")
return [s[0] for s in sorted_stocks[:n]]
# ── BACKTESTING ENGINE ───────────────────────────────────────────────
def backtest_momentum_strategy(
start_date="2022-01-01",
end_date="2024-12-31",
initial_capital=50000
):
"""Backtest the momentum strategy on historical data"""
capital = initial_capital
portfolio = {} # {symbol: {qty, entry_price, entry_date}}
trades = []
equity_curve = [capital]
# Weekly rebalancing simulation
start = pd.Timestamp(start_date)
end = pd.Timestamp(end_date)
rebalance_dates = pd.date_range(start=start, end=end, freq='W-MON')
for rebalance_date in rebalance_dates:
logger.info(f"\nRebalancing on {rebalance_date.date()}")
# Get momentum scores for this date
scores = {}
for symbol in NIFTY50_STOCKS:
try:
ticker = yf.Ticker(f"{symbol}.NS")
hist = ticker.history(start=rebalance_date - timedelta(days=60),
end=rebalance_date)
if len(hist) >= LOOKBACK_DAYS:
price_mom = (hist['Close'].iloc[-1] / hist['Close'].iloc[-LOOKBACK_DAYS] - 1) * 100
scores[symbol] = price_mom
except:
pass
target_stocks = sorted(scores, key=scores.get, reverse=True)[:MAX_POSITIONS]
# Exit positions not in new top stocks
for symbol in list(portfolio.keys()):
if symbol not in target_stocks:
try:
ticker = yf.Ticker(f"{symbol}.NS")
hist = ticker.history(start=rebalance_date, end=rebalance_date + timedelta(days=3))
exit_price = hist['Open'].iloc[0] if not hist.empty else portfolio[symbol]['entry_price']
qty = portfolio[symbol]['qty']
entry_price = portfolio[symbol]['entry_price']
pnl = (exit_price - entry_price) * qty
capital += exit_price * qty
trades.append({
"symbol": symbol,
"type": "EXIT",
"entry": entry_price,
"exit": exit_price,
"qty": qty,
"pnl": pnl,
"return_pct": (exit_price/entry_price - 1) * 100
})
del portfolio[symbol]
logger.info(f" EXIT {symbol}: PnL = ₹{pnl:.0f}")
except Exception as e:
logger.error(f"Error exiting {symbol}: {e}")
# Enter new positions
capital_per_stock = (capital / MAX_POSITIONS) * 0.95 # 5% cash buffer
for symbol in target_stocks:
if symbol not in portfolio:
try:
ticker = yf.Ticker(f"{symbol}.NS")
hist = ticker.history(start=rebalance_date, end=rebalance_date + timedelta(days=3))
if hist.empty:
continue
entry_price = hist['Open'].iloc[0]
qty = int(capital_per_stock / entry_price)
if qty > 0 and entry_price * qty <= capital:
capital -= entry_price * qty
portfolio[symbol] = {
"qty": qty,
"entry_price": entry_price,
"entry_date": rebalance_date
}
logger.info(f" BUY {symbol}: {qty} @ ₹{entry_price:.2f}")
except Exception as e:
logger.error(f"Error entering {symbol}: {e}")
# Calculate portfolio value
portfolio_value = capital
for symbol, pos in portfolio.items():
try:
ticker = yf.Ticker(f"{symbol}.NS")
hist = ticker.history(start=rebalance_date, end=rebalance_date + timedelta(days=3))
if not hist.empty:
portfolio_value += hist['Close'].iloc[-1] * pos['qty']
except:
portfolio_value += pos['entry_price'] * pos['qty']
equity_curve.append(portfolio_value)
# Performance metrics
total_return = (equity_curve[-1] / initial_capital - 1) * 100
max_drawdown = min([(equity_curve[i] / max(equity_curve[:i+1]) - 1) * 100
for i in range(1, len(equity_curve))], default=0)
winning_trades = [t for t in trades if t['pnl'] > 0]
win_rate = len(winning_trades) / len(trades) * 100 if trades else 0
print("\n" + "="*60)
print("BACKTEST RESULTS")
print("="*60)
print(f"Period: {start_date} to {end_date}")
print(f"Initial Capital: ₹{initial_capital:,}")
print(f"Final Capital: ₹{equity_curve[-1]:,.0f}")
print(f"Total Return: {total_return:.1f}%")
print(f"Max Drawdown: {max_drawdown:.1f}%")
print(f"Total Trades: {len(trades)}")
print(f"Win Rate: {win_rate:.1f}%")
return {"total_return": total_return, "max_drawdown": max_drawdown,
"trades": trades, "equity_curve": equity_curve}
# ── LIVE TRADING (Zerodha) ───────────────────────────────────────────
class ZerodhaTrader:
def __init__(self, api_key, api_secret, access_token):
self.kite = KiteConnect(api_key=api_key)
self.kite.set_access_token(access_token)
logger.info("Zerodha connected successfully")
def place_market_order(self, symbol, qty, side):
"""Place market order — MIS (intraday) or CNC (delivery)"""
try:
order_id = self.kite.place_order(
variety=KiteConnect.VARIETY_REGULAR,
exchange=KiteConnect.EXCHANGE_NSE,
tradingsymbol=symbol,
transaction_type=KiteConnect.TRANSACTION_TYPE_BUY if side == 'BUY'
else KiteConnect.TRANSACTION_TYPE_SELL,
quantity=qty,
product=KiteConnect.PRODUCT_CNC, # CNC = delivery, MIS = intraday
order_type=KiteConnect.ORDER_TYPE_MARKET
)
logger.info(f"Order placed: {side} {qty} {symbol} — Order ID: {order_id}")
return order_id
except Exception as e:
logger.error(f"Order failed: {e}")
return None
def rebalance_portfolio(self, target_stocks):
"""Rebalance to target stocks"""
current_positions = {
pos['tradingsymbol']: pos['quantity']
for pos in self.kite.positions()['net']
if pos['quantity'] > 0
}
# Exit positions not in target
for symbol, qty in current_positions.items():
if symbol not in target_stocks:
self.place_market_order(symbol, qty, 'SELL')
time.sleep(1) # Rate limiting
# Enter new positions
funds = self.kite.margins()['equity']['available']['live_balance']
capital_per_stock = (funds / MAX_POSITIONS) * 0.95
for symbol in target_stocks:
if symbol not in current_positions:
ltp = self.kite.ltp(f"NSE:{symbol}")[f"NSE:{symbol}"]['last_price']
qty = int(capital_per_stock / ltp)
if qty > 0:
self.place_market_order(symbol, qty, 'BUY')
time.sleep(1)
# ── RUN ──────────────────────────────────────────────────────────────
if __name__ == "__main__":
# 1. Run backtest first
results = backtest_momentum_strategy(
start_date="2022-01-01",
end_date="2024-12-31",
initial_capital=50000
)
# 2. For live trading, uncomment and add your Zerodha credentials:
# trader = ZerodhaTrader(
# api_key="your_api_key",
# api_secret="your_api_secret",
# access_token="your_access_token" # Get fresh each day
# )
# top_stocks = get_top_momentum_stocks(MAX_POSITIONS)
# trader.rebalance_portfolio(top_stocks)
# DISCLAIMER: This is for educational purposes only.
# Past performance does not guarantee future results.
# Trading involves substantial risk of loss.