Portfolio Optimization with Python: Modern Portfolio Theory for Indian Markets
Master portfolio optimization techniques using Python. Learn mean-variance optimization, risk parity, and Black-Litterman models tailored for NSE/BSE.
You have ₹10 lakhs to invest across Nifty 50 stocks. How do you allocate capital to maximize returns while managing risk? Should you equal-weight all 50 stocks? Concentrate in your highest-conviction picks? Or use mathematical optimization to find the “efficient frontier”?
This comprehensive guide teaches you portfolio optimization techniques using Python, specifically adapted for Indian market realities including transaction costs, liquidity constraints, and SEBI regulations.
Part 1: Modern Portfolio Theory Foundations
The Core Concept
Harry Markowitz’s insight: Don’t just look at individual stock returns—consider how they move together (correlation).
Key Metrics:
import pandas as pd
import numpy as np
import yfinance as yf
from datetime import datetime, timedelta
class PortfolioAnalyzer:
"""
Portfolio analysis for Indian markets
"""
def __init__(self, symbols: list, start_date: str, end_date: str):
self.symbols = [f"{s}.NS" for s in symbols] # Add .NS for NSE
self.start_date = start_date
self.end_date = end_date
self.data = None
self.returns = None
def fetch_data(self):
"""Fetch historical data"""
self.data = yf.download(
self.symbols,
start=self.start_date,
end=self.end_date,
progress=False
)['Adj Close']
# Calculate returns
self.returns = self.data.pct_change().dropna()
return self.data
def calculate_portfolio_metrics(self, weights: np.array) -> dict:
"""
Calculate portfolio return, volatility, and Sharpe ratio
Weights: numpy array of portfolio weights (should sum to 1)
"""
# Annualize returns and covariance (252 trading days in India)
mean_returns = self.returns.mean() * 252
cov_matrix = self.returns.cov() * 252
# Portfolio return
portfolio_return = np.dot(weights, mean_returns)
# Portfolio volatility (risk)
portfolio_variance = np.dot(weights.T, np.dot(cov_matrix, weights))
portfolio_std = np.sqrt(portfolio_variance)
# Sharpe ratio (assuming 6% risk-free rate for India)
risk_free_rate = 0.06
sharpe_ratio = (portfolio_return - risk_free_rate) / portfolio_std
return {
'return': portfolio_return,
'volatility': portfolio_std,
'sharpe_ratio': sharpe_ratio,
'weights': weights
}
def get_correlation_matrix(self) -> pd.DataFrame:
"""Calculate correlation matrix"""
return self.returns.corr()
def get_covariance_matrix(self) -> pd.DataFrame:
"""Calculate annualized covariance matrix"""
return self.returns.cov() * 252
# Example: Analyze a portfolio of 5 stocks
symbols = ['RELIANCE', 'TCS', 'HDFCBANK', 'INFY', 'ICICIBANK']
analyzer = PortfolioAnalyzer(
symbols=symbols,
start_date='2023-01-01',
end_date='2024-12-31'
)
data = analyzer.fetch_data()
# Equal-weighted portfolio
equal_weights = np.array([0.2, 0.2, 0.2, 0.2, 0.2])
metrics = analyzer.calculate_portfolio_metrics(equal_weights)
print(f"Portfolio Return: {metrics['return']:.2%}")
print(f"Portfolio Volatility: {metrics['volatility']:.2%}")
print(f"Sharpe Ratio: {metrics['sharpe_ratio']:.2f}")
Visualization: Correlation Heatmap
import matplotlib.pyplot as plt
import seaborn as sns
def plot_correlation_matrix(analyzer):
"""
Visualize correlation between stocks
"""
corr = analyzer.get_correlation_matrix()
plt.figure(figsize=(10, 8))
sns.heatmap(
corr,
annot=True,
cmap='coolwarm',
center=0,
fmt='.2f',
square=True,
linewidths=1
)
plt.title('Stock Correlation Matrix', fontsize=14, fontweight='bold')
plt.tight_layout()
plt.show()
plot_correlation_matrix(analyzer)
Part 2: Mean-Variance Optimization
Finding the Efficient Frontier
from scipy.optimize import minimize
import matplotlib.pyplot as plt
class PortfolioOptimizer:
"""
Portfolio optimization using mean-variance framework
"""
def __init__(self, returns: pd.DataFrame):
self.returns = returns
self.num_assets = len(returns.columns)
self.mean_returns = returns.mean() * 252
self.cov_matrix = returns.cov() * 252
def portfolio_stats(self, weights):
"""Calculate portfolio statistics"""
returns = np.dot(weights, self.mean_returns)
std = np.sqrt(np.dot(weights.T, np.dot(self.cov_matrix, weights)))
return returns, std
def negative_sharpe(self, weights, risk_free_rate=0.06):
"""
Negative Sharpe ratio (for minimization)
"""
returns, std = self.portfolio_stats(weights)
sharpe = (returns - risk_free_rate) / std
return -sharpe # Negative because we minimize
def portfolio_variance(self, weights):
"""Calculate portfolio variance"""
return np.dot(weights.T, np.dot(self.cov_matrix, weights))
def optimize_sharpe(self, constraints=None) -> dict:
"""
Find portfolio with maximum Sharpe ratio
Constraints:
- Weights sum to 1
- No short selling (weights >= 0)
- Optional: Max position size, sector limits, etc.
"""
# Initial guess: equal weights
init_guess = np.array([1 / self.num_assets] * self.num_assets)
# Constraints
constraints_list = [
{'type': 'eq', 'fun': lambda x: np.sum(x) - 1} # Weights sum to 1
]
if constraints:
constraints_list.extend(constraints)
# Bounds: 0 <= weight <= 1 (no short selling)
bounds = tuple((0, 1) for _ in range(self.num_assets))
# Optimize
result = minimize(
self.negative_sharpe,
init_guess,
method='SLSQP',
bounds=bounds,
constraints=constraints_list
)
# Extract optimal weights
optimal_weights = result.x
# Calculate metrics
portfolio_return, portfolio_std = self.portfolio_stats(optimal_weights)
sharpe = (portfolio_return - 0.06) / portfolio_std
return {
'weights': optimal_weights,
'return': portfolio_return,
'volatility': portfolio_std,
'sharpe_ratio': sharpe
}
def optimize_min_variance(self) -> dict:
"""
Find minimum variance portfolio
"""
init_guess = np.array([1 / self.num_assets] * self.num_assets)
constraints = [
{'type': 'eq', 'fun': lambda x: np.sum(x) - 1}
]
bounds = tuple((0, 1) for _ in range(self.num_assets))
result = minimize(
self.portfolio_variance,
init_guess,
method='SLSQP',
bounds=bounds,
constraints=constraints
)
optimal_weights = result.x
portfolio_return, portfolio_std = self.portfolio_stats(optimal_weights)
sharpe = (portfolio_return - 0.06) / portfolio_std
return {
'weights': optimal_weights,
'return': portfolio_return,
'volatility': portfolio_std,
'sharpe_ratio': sharpe
}
def efficient_frontier(self, num_portfolios=100):
"""
Generate efficient frontier
"""
min_var_port = self.optimize_min_variance()
max_sharpe_port = self.optimize_sharpe()
# Range of target returns
min_return = min_var_port['return']
max_return = self.mean_returns.max()
target_returns = np.linspace(min_return, max_return, num_portfolios)
efficient_portfolios = []
for target_return in target_returns:
# Constraint: Portfolio return = target
constraints = [
{'type': 'eq', 'fun': lambda x: np.sum(x) - 1},
{
'type': 'eq',
'fun': lambda x: self.portfolio_stats(x)[0] - target_return
}
]
bounds = tuple((0, 1) for _ in range(self.num_assets))
init_guess = np.array([1 / self.num_assets] * self.num_assets)
try:
result = minimize(
self.portfolio_variance,
init_guess,
method='SLSQP',
bounds=bounds,
constraints=constraints
)
if result.success:
weights = result.x
returns, std = self.portfolio_stats(weights)
efficient_portfolios.append({
'return': returns,
'volatility': std,
'weights': weights
})
except:
pass
return efficient_portfolios
# Usage
optimizer = PortfolioOptimizer(analyzer.returns)
# Find maximum Sharpe ratio portfolio
max_sharpe = optimizer.optimize_sharpe()
print("\n=== Maximum Sharpe Ratio Portfolio ===")
for i, symbol in enumerate(symbols):
if max_sharpe['weights'][i] > 0.01: # Show only significant allocations
print(f"{symbol}: {max_sharpe['weights'][i]:.2%}")
print(f"\nExpected Return: {max_sharpe['return']:.2%}")
print(f"Volatility: {max_sharpe['volatility']:.2%}")
print(f"Sharpe Ratio: {max_sharpe['sharpe_ratio']:.2f}")
# Find minimum variance portfolio
min_var = optimizer.optimize_min_variance()
print("\n=== Minimum Variance Portfolio ===")
for i, symbol in enumerate(symbols):
if min_var['weights'][i] > 0.01:
print(f"{symbol}: {min_var['weights'][i]:.2%}")
print(f"\nExpected Return: {min_var['return']:.2%}")
print(f"Volatility: {min_var['volatility']:.2%}")
print(f"Sharpe Ratio: {min_var['sharpe_ratio']:.2f}")
Visualizing the Efficient Frontier
def plot_efficient_frontier(optimizer, max_sharpe, min_var):
"""
Plot efficient frontier with key portfolios marked
"""
# Generate efficient frontier
efficient_portfolios = optimizer.efficient_frontier(num_portfolios=100)
if not efficient_portfolios:
print("Could not generate efficient frontier")
return
# Extract data
returns = [p['return'] for p in efficient_portfolios]
volatilities = [p['volatility'] for p in efficient_portfolios]
# Plot
plt.figure(figsize=(12, 8))
# Efficient frontier
plt.plot(volatilities, returns, 'b-', linewidth=2, label='Efficient Frontier')
# Maximum Sharpe ratio
plt.scatter(
max_sharpe['volatility'],
max_sharpe['return'],
c='red',
marker='*',
s=500,
label=f"Max Sharpe ({max_sharpe['sharpe_ratio']:.2f})"
)
# Minimum variance
plt.scatter(
min_var['volatility'],
min_var['return'],
c='green',
marker='*',
s=500,
label=f"Min Variance"
)
# Equal weight portfolio (for comparison)
equal_weights = np.array([1 / optimizer.num_assets] * optimizer.num_assets)
eq_return, eq_vol = optimizer.portfolio_stats(equal_weights)
plt.scatter(
eq_vol,
eq_return,
c='orange',
marker='o',
s=200,
label='Equal Weight'
)
# Individual stocks
for i, symbol in enumerate(symbols):
stock_return = optimizer.mean_returns.iloc[i]
stock_vol = np.sqrt(optimizer.cov_matrix.iloc[i, i])
plt.scatter(stock_vol, stock_return, c='gray', alpha=0.5, s=50)
plt.annotate(
symbol,
(stock_vol, stock_return),
fontsize=8,
alpha=0.7
)
plt.xlabel('Volatility (Risk)', fontsize=12)
plt.ylabel('Expected Return', fontsize=12)
plt.title('Efficient Frontier - Indian Stocks', fontsize=14, fontweight='bold')
plt.legend()
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
plot_efficient_frontier(optimizer, max_sharpe, min_var)
Part 3: Practical Constraints
Indian Market Realities
class RealWorldPortfolioOptimizer(PortfolioOptimizer):
"""
Portfolio optimizer with real-world constraints
"""
def optimize_with_constraints(
self,
min_position: float = 0.05, # Minimum 5% per stock
max_position: float = 0.25, # Maximum 25% per stock
max_stocks: int = 10, # Maximum number of positions
sector_limits: dict = None # Sector exposure limits
) -> dict:
"""
Optimize with practical constraints
"""
init_guess = np.array([1 / self.num_assets] * self.num_assets)
# Basic constraints
constraints = [
{'type': 'eq', 'fun': lambda x: np.sum(x) - 1} # Sum to 1
]
# Position size constraints
# If weight > 0, it must be >= min_position
# This is tricky to implement, so we use bounds instead
bounds = tuple((0, max_position) for _ in range(self.num_assets))
result = minimize(
self.negative_sharpe,
init_guess,
method='SLSQP',
bounds=bounds,
constraints=constraints
)
optimal_weights = result.x
# Post-process: Apply minimum position constraint
# Zero out positions below threshold
optimal_weights[optimal_weights < min_position] = 0
# Limit number of stocks
if np.sum(optimal_weights > 0) > max_stocks:
# Keep only top N positions
top_indices = np.argsort(optimal_weights)[-max_stocks:]
mask = np.zeros_like(optimal_weights, dtype=bool)
mask[top_indices] = True
optimal_weights[~mask] = 0
# Renormalize
optimal_weights = optimal_weights / np.sum(optimal_weights)
# Calculate metrics
portfolio_return, portfolio_std = self.portfolio_stats(optimal_weights)
sharpe = (portfolio_return - 0.06) / portfolio_std
return {
'weights': optimal_weights,
'return': portfolio_return,
'volatility': portfolio_std,
'sharpe_ratio': sharpe,
'num_positions': np.sum(optimal_weights > 0)
}
# Usage
real_optimizer = RealWorldPortfolioOptimizer(analyzer.returns)
constrained_portfolio = real_optimizer.optimize_with_constraints(
min_position=0.10, # Minimum 10% per stock
max_position=0.30, # Maximum 30% per stock
max_stocks=5 # Maximum 5 stocks
)
print("\n=== Constrained Portfolio ===")
for i, symbol in enumerate(symbols):
if constrained_portfolio['weights'][i] > 0:
print(f"{symbol}: {constrained_portfolio['weights'][i]:.2%}")
print(f"\nNumber of Positions: {constrained_portfolio['num_positions']}")
print(f"Expected Return: {constrained_portfolio['return']:.2%}")
print(f"Volatility: {constrained_portfolio['volatility']:.2%}")
print(f"Sharpe Ratio: {constrained_portfolio['sharpe_ratio']:.2f}")
Transaction Cost Aware Optimization
class TransactionCostOptimizer(PortfolioOptimizer):
"""
Optimize considering transaction costs
"""
def __init__(
self,
returns: pd.DataFrame,
current_weights: np.array,
transaction_cost_pct: float = 0.001 # 0.1% per trade
):
super().__init__(returns)
self.current_weights = current_weights
self.transaction_cost_pct = transaction_cost_pct
def calculate_turnover_cost(self, new_weights: np.array) -> float:
"""
Calculate cost of rebalancing
"""
# Turnover = sum of absolute weight changes
turnover = np.sum(np.abs(new_weights - self.current_weights))
# Cost = turnover * transaction cost %
cost = turnover * self.transaction_cost_pct
return cost
def net_return_after_costs(self, weights):
"""
Calculate expected return minus rebalancing costs
"""
gross_return = np.dot(weights, self.mean_returns)
rebalancing_cost = self.calculate_turnover_cost(weights)
net_return = gross_return - rebalancing_cost
return net_return
def optimize_with_transaction_costs(self) -> dict:
"""
Optimize considering transaction costs
"""
init_guess = self.current_weights
constraints = [
{'type': 'eq', 'fun': lambda x: np.sum(x) - 1}
]
bounds = tuple((0, 1) for _ in range(self.num_assets))
def objective(weights):
"""
Objective: Maximize Sharpe ratio considering transaction costs
"""
net_return = self.net_return_after_costs(weights)
std = np.sqrt(np.dot(weights.T, np.dot(self.cov_matrix, weights)))
sharpe = (net_return - 0.06) / std
return -sharpe
result = minimize(
objective,
init_guess,
method='SLSQP',
bounds=bounds,
constraints=constraints
)
optimal_weights = result.x
# Calculate metrics
gross_return, std = self.portfolio_stats(optimal_weights)
rebalancing_cost = self.calculate_turnover_cost(optimal_weights)
net_return = gross_return - rebalancing_cost
sharpe = (net_return - 0.06) / std
# Turnover percentage
turnover_pct = np.sum(np.abs(optimal_weights - self.current_weights))
return {
'weights': optimal_weights,
'gross_return': gross_return,
'rebalancing_cost': rebalancing_cost,
'net_return': net_return,
'volatility': std,
'sharpe_ratio': sharpe,
'turnover_pct': turnover_pct
}
# Example: Current portfolio vs optimal rebalancing
current_weights = np.array([0.15, 0.25, 0.30, 0.20, 0.10])
cost_optimizer = TransactionCostOptimizer(
returns=analyzer.returns,
current_weights=current_weights,
transaction_cost_pct=0.001 # 0.1% per trade
)
optimal_rebalance = cost_optimizer.optimize_with_transaction_costs()
print("\n=== Rebalancing Analysis ===")
print("\nCurrent Allocation:")
for i, symbol in enumerate(symbols):
print(f"{symbol}: {current_weights[i]:.2%}")
print("\nOptimal Allocation:")
for i, symbol in enumerate(symbols):
change = optimal_rebalance['weights'][i] - current_weights[i]
arrow = "↑" if change > 0 else "↓" if change < 0 else "→"
print(f"{symbol}: {optimal_rebalance['weights'][i]:.2%} ({arrow} {abs(change):.2%})")
print(f"\nGross Return: {optimal_rebalance['gross_return']:.2%}")
print(f"Rebalancing Cost: {optimal_rebalance['rebalancing_cost']:.2%}")
print(f"Net Return: {optimal_rebalance['net_return']:.2%}")
print(f"Turnover: {optimal_rebalance['turnover_pct']:.2%}")
print(f"Sharpe Ratio: {optimal_rebalance['sharpe_ratio']:.2f}")
Part 4: Risk Parity
Equal risk contribution from each asset.
class RiskParityOptimizer:
"""
Risk parity portfolio optimization
All assets contribute equally to portfolio risk
"""
def __init__(self, returns: pd.DataFrame):
self.returns = returns
self.num_assets = len(returns.columns)
self.cov_matrix = returns.cov() * 252
def calculate_risk_contribution(self, weights: np.array) -> np.array:
"""
Calculate risk contribution of each asset
"""
portfolio_variance = np.dot(weights.T, np.dot(self.cov_matrix, weights))
portfolio_volatility = np.sqrt(portfolio_variance)
# Marginal contribution to risk
marginal_contrib = np.dot(self.cov_matrix, weights) / portfolio_volatility
# Risk contribution = weight * marginal contribution
risk_contrib = weights * marginal_contrib
return risk_contrib
def risk_parity_objective(self, weights: np.array) -> float:
"""
Objective: Minimize variance of risk contributions
"""
risk_contrib = self.calculate_risk_contribution(weights)
# Target: equal risk contribution
target = np.ones(self.num_assets) / self.num_assets
# Minimize sum of squared deviations
return np.sum((risk_contrib - target) ** 2)
def optimize(self) -> dict:
"""
Find risk parity portfolio
"""
init_guess = np.array([1 / self.num_assets] * self.num_assets)
constraints = [
{'type': 'eq', 'fun': lambda x: np.sum(x) - 1}
]
bounds = tuple((0.01, 1) for _ in range(self.num_assets))
result = minimize(
self.risk_parity_objective,
init_guess,
method='SLSQP',
bounds=bounds,
constraints=constraints
)
optimal_weights = result.x
# Calculate risk contributions
risk_contrib = self.calculate_risk_contribution(optimal_weights)
risk_contrib_pct = risk_contrib / risk_contrib.sum()
return {
'weights': optimal_weights,
'risk_contributions': risk_contrib_pct
}
# Usage
rp_optimizer = RiskParityOptimizer(analyzer.returns)
risk_parity_portfolio = rp_optimizer.optimize()
print("\n=== Risk Parity Portfolio ===")
for i, symbol in enumerate(symbols):
print(f"{symbol}:")
print(f" Weight: {risk_parity_portfolio['weights'][i]:.2%}")
print(f" Risk Contribution: {risk_parity_portfolio['risk_contributions'][i]:.2%}")
Conclusion
Portfolio optimization transforms gut-feel allocation into mathematically rigorous strategies:
- Mean-Variance Optimization: Find efficient frontier, maximize Sharpe ratio
- Practical Constraints: Position limits, transaction costs, turnover minimization
- Risk Parity: Equal risk contribution for diversified exposure
- Rebalancing: Balance improvement vs. transaction costs
Start with maximum Sharpe portfolio, add constraints as needed, backtest thoroughly.
Ready to optimize your portfolio? Contact us for personalized portfolio optimization services.