USER
Adjust Return Calculation for Long-Term Focus: Switch from annualized metrics to multi-year compounding to better reflect long-term growth potential.
Incorporate Economic Cycles: Implement features or conditions that simulate economic cycles (e.g., growth, recession) to observe how portfolios perform in different market phases.
Enhanced Downside Risk Measures: Replace or complement the Sortino ratio with additional downside risk metrics, like Conditional Value at Risk (CVaR), which is useful for assessing tail risk over extended periods.
apply the above improvement to the system and return me the full enhanced code, no need any explnation:
# Install required packages
!pip install yfinance pandas numpy beautifulsoup4 requests tqdm deap matplotlib
# Import necessary libraries
import yfinance as yf
import pandas as pd
import numpy as np
import random
from bs4 import BeautifulSoup
import requests
from tqdm import tqdm
import sys
import warnings
import multiprocessing
import matplotlib.pyplot as plt
from deap import base, creator, tools, algorithms
# Suppress warnings for cleaner output
warnings.filterwarnings("ignore")
# Set random seed for reproducibility
RANDOM_SEED = 42
random.seed(RANDOM_SEED)
np.random.seed(RANDOM_SEED)
# Function to fetch S&P 500 tickers from Wikipedia
def get_sp500_tickers():
url = 'https://en.wikipedia.org/wiki/List_of_S%26P_500_companies'
response = requests.get(url)
if response.status_code != 200:
raise Exception("Failed to fetch S&P 500 tickers from Wikipedia.")
soup = BeautifulSoup(response.text, "lxml")
table = soup.find('table', {'id': 'constituents'})
tickers = []
for row in table.findAll('tr')[1:]:
ticker = row.findAll('td')[0].text.strip()
# Replace '.' with '-' for tickers like BRK.B
ticker = ticker.replace('.', '-')
tickers.append(ticker)
return tickers
# Function to calculate Maximum Drawdown
def max_drawdown(series):
try:
roll_max = series.cummax()
drawdown = (series - roll_max) / roll_max
return drawdown.min()
except:
return -0.0
# Function to calculate Sortino Ratio
def sortino_ratio(returns, target=0):
downside = returns[returns < target]
if len(downside) == 0:
return 0 # No downside risk; assign a neutral value
expected_return = returns.mean() * 252
downside_risk = np.sqrt((downside ** 2).mean()) * np.sqrt(252)
return (expected_return - target) / downside_risk if downside_risk != 0 else 0
# Function to generate portfolio weights within constraints
def generate_weights(n, lower=0.10, upper=0.20, max_attempts=1000):
"""
Generates a list of weights that sum to 1.0 with each weight between lower and upper bounds.
"""
for _ in range(max_attempts):
weights = np.random.uniform(lower, upper, n)
weights /= weights.sum()
if all(lower <= w <= upper for w in weights):
return weights
raise ValueError("Unable to generate weights within constraints after multiple attempts.")
# Parameters
START_DATE = '2000-01-01'
END_DATE = '2024-10-30'
NUM_ASSETS = 8 # Number of stocks in the portfolio
LOWER_WEIGHT = 0.10
UPPER_WEIGHT = 0.20
RISK_FREE_RATE = 0.017 # Approximate risk-free rate (e.g., 10-year US Treasury yield)
# Step 1: Get S&P 500 tickers
print("Fetching S&P 500 tickers...")
tickers = get_sp500_tickers()
print(f"Number of tickers fetched: {len(tickers)}")
# Step 2: Download historical data with dynamic start date
print("Downloading historical data...")
failed_tickers = []
successful_tickers = []
data = pd.DataFrame()
ticker_first_dates = {}
for ticker in tqdm(tickers, desc="Downloading tickers"):
try:
df = yf.download(ticker, start=START_DATE, end=END_DATE, progress=False)['Adj Close']
if df.empty:
failed_tickers.append(ticker)
else:
data[ticker] = df
successful_tickers.append(ticker)
ticker_first_dates[ticker] = df.first_valid_index()
except Exception as e:
failed_tickers.append(ticker)
print(f"\nTotal tickers with successful data download: {len(successful_tickers)}")
print(f"Total tickers failed to download: {len(failed_tickers)}")
if failed_tickers:
print(f"Failed tickers: {failed_tickers}")
# Determine the earliest start date among all successful tickers
if ticker_first_dates:
earliest_date = min(ticker_first_dates.values())
print(f"Earliest available data starts from: {earliest_date.date()}")
# Adjust the dataset to start from the earliest_date
data = data.loc[earliest_date:]
# Forward fill to propagate last valid observation
data.fillna(method='ffill', inplace=True)
# Backward fill to handle any remaining NaNs at the start
data.fillna(method='bfill', inplace=True)
print(f"Data shape after adjusting start date and dropping incomplete data: {data.shape}")
else:
print("No data available after processing tickers.")
sys.exit()
# If there are not enough tickers, exit
if len(data.columns) < NUM_ASSETS:
print("Not enough tickers with complete data to form a portfolio.")
sys.exit()
# Step 3: Calculate daily returns
returns = data.pct_change().dropna()
# Step 4: Split data into Training (70%), Validation (15%), Testing (15%)
split_date_1 = int(len(returns) * 0.70)
split_date_2 = int(len(returns) * 0.85)
train_returns = returns.iloc[:split_date_1]
validation_returns = returns.iloc[split_date_1:split_date_2]
test_returns = returns.iloc[split_date_2:]
print(f"Training set: {train_returns.shape[0]} days")
print(f"Validation set: {validation_returns.shape[0]} days")
print(f"Testing set: {test_returns.shape[0]} days")
# Calculate annualized metrics for each set
def calculate_metrics(returns_set):
ann_return = returns_set.mean() * 252
ann_volatility = returns_set.std() * np.sqrt(252)
cov = returns_set.cov() * 252
return ann_return, ann_volatility, cov
train_ann_return, train_ann_volatility, train_cov_matrix = calculate_metrics(train_returns)
validation_ann_return, validation_ann_volatility, validation_cov_matrix = calculate_metrics(validation_returns)
test_ann_return, test_ann_volatility, test_cov_matrix = calculate_metrics(test_returns)
# Define Genetic Algorithm parameters
POPULATION_SIZE = 500 # Adjusted for memory constraints
P_CROSSOVER = 0.8 # Probability for crossover
P_MUTATION = 0.2 # Probability for mutating an individual
MAX_GENERATIONS = 100 # Increased number of generations for better convergence
HALL_OF_FAME_SIZE = 1 # Number of best individuals to keep
# Define the weights for each optimization objective
WEIGHT_SHARPE = 0.25
WEIGHT_MAX_DRAWDOWN = 0.20
WEIGHT_ANNUAL_RETURN = 0.25
WEIGHT_VOLATILITY = 0.15
WEIGHT_SORTINO = 0.15
# Define the evaluation (fitness) function
def evaluate_portfolio(individual):
selected_tickers = individual[:NUM_ASSETS]
weights = individual[NUM_ASSETS:]
# Ensure all selected tickers are unique
if len(set(selected_tickers)) != NUM_ASSETS:
return -np.inf, # Invalid individual
# Ensure weights sum to 1 and are within constraints
weights = np.array(weights)
if not np.isclose(weights.sum(), 1.0):
return -np.inf,
if not all(LOWER_WEIGHT <= w <= UPPER_WEIGHT for w in weights):
return -np.inf,
# Check if all tickers are present in the covariance matrix
if not all(ticker in train_cov_matrix.columns for ticker in selected_tickers):
return -np.inf, # Invalid individual
# Calculate portfolio return and volatility using Training Set
try:
portfolio_return_train = np.dot(weights, train_ann_return[selected_tickers])
portfolio_volatility_train = np.sqrt(np.dot(weights, np.dot(train_cov_matrix.loc[selected_tickers, selected_tickers].values, weights)))
except KeyError as e:
# In case tickers are not found in covariance matrix
return -np.inf,
if portfolio_volatility_train == 0:
return -np.inf,
# Calculate Sharpe Ratio for Training
sharpe_ratio_train = (portfolio_return_train - RISK_FREE_RATE) / portfolio_volatility_train
# Calculate Sortino Ratio for Training
portfolio_sortino_train = sortino_ratio(train_returns[selected_tickers].dot(weights))
# Calculate portfolio cumulative returns for Max Drawdown using Training Set
portfolio_cum_returns_train = (1 + train_returns[selected_tickers].dot(weights)).cumprod()
portfolio_max_drawdown_train = max_drawdown(portfolio_cum_returns_train)
# Calculate Validation Metrics
try:
portfolio_return_val = np.dot(weights, validation_ann_return[selected_tickers])
portfolio_volatility_val = np.sqrt(np.dot(weights, np.dot(validation_cov_matrix.loc[selected_tickers, selected_tickers].values, weights)))
except KeyError as e:
return -np.inf,
if portfolio_volatility_val == 0:
return -np.inf,
# Calculate Sharpe Ratio for Validation
sharpe_ratio_val = (portfolio_return_val - RISK_FREE_RATE) / portfolio_volatility_val if portfolio_volatility_val != 0 else 0
# Calculate Sortino Ratio for Validation
portfolio_sortino_val = sortino_ratio(validation_returns[selected_tickers].dot(weights))
# Calculate portfolio cumulative returns for Max Drawdown using Validation Set
portfolio_cum_returns_val = (1 + validation_returns[selected_tickers].dot(weights)).cumprod()
portfolio_max_drawdown_val = max_drawdown(portfolio_cum_returns_val)
# Combine the metrics into a single fitness score
# Weighted average of training and validation scores
fitness_train = (WEIGHT_SHARPE * sharpe_ratio_train) \
- (WEIGHT_MAX_DRAWDOWN * portfolio_max_drawdown_train) \
+ (WEIGHT_ANNUAL_RETURN * portfolio_return_train) \
- (WEIGHT_VOLATILITY * portfolio_volatility_train) \
+ (WEIGHT_SORTINO * portfolio_sortino_train)
fitness_val = (WEIGHT_SHARPE * sharpe_ratio_val) \
- (WEIGHT_MAX_DRAWDOWN * portfolio_max_drawdown_val) \
+ (WEIGHT_ANNUAL_RETURN * portfolio_return_val) \
- (WEIGHT_VOLATILITY * portfolio_volatility_val) \
+ (WEIGHT_SORTINO * portfolio_sortino_val)
# Combine Training and Validation Fitness
fitness = (fitness_train + fitness_val) / 2
return fitness,
# Create the DEAP framework
creator.create("FitnessMax", base.Fitness, weights=(1.0,)) # Single objective, maximize
creator.create("Individual", list, fitness=creator.FitnessMax)
toolbox = base.Toolbox()
# Attribute generators
# Selection part: choosing 8 unique tickers
def select_tickers():
return random.sample(successful_tickers, NUM_ASSETS)
# Allocation part: generating 8 weights between 10% and 20% that sum to 1
def allocate_weights():
return generate_weights(NUM_ASSETS, LOWER_WEIGHT, UPPER_WEIGHT).tolist()
# Define how each individual is created
def create_individual():
tickers = select_tickers()
weights = allocate_weights()
return creator.Individual(tickers + weights)
toolbox.register("individual", create_individual)
toolbox.register("population", tools.initRepeat, list, toolbox.individual)
# Custom Crossover: Ensures tickers remain unique and weights are within constraints
def cxTickersWeights(ind1, ind2):
# Separate tickers and weights
tickers1, weights1 = ind1[:NUM_ASSETS], ind1[NUM_ASSETS:]
tickers2, weights2 = ind2[:NUM_ASSETS], ind2[NUM_ASSETS:]
# Crossover tickers using one-point crossover
cx_point = random.randint(1, NUM_ASSETS -1)
new_tickers1 = tickers1[:cx_point] + [ticker for ticker in tickers2[cx_point:] if ticker not in tickers1[:cx_point]]
new_tickers2 = tickers2[:cx_point] + [ticker for ticker in tickers1[cx_point:] if ticker not in tickers2[:cx_point]]
# Fill the remaining tickers to maintain NUM_ASSETS
available_tickers1 = list(set(successful_tickers) - set(new_tickers1))
available_tickers2 = list(set(successful_tickers) - set(new_tickers2))
while len(new_tickers1) < NUM_ASSETS:
new_tickers1.append(random.choice(available_tickers1))
available_tickers1.remove(new_tickers1[-1])
while len(new_tickers2) < NUM_ASSETS:
new_tickers2.append(random.choice(available_tickers2))
available_tickers2.remove(new_tickers2[-1])
# Assign the new tickers
ind1[:NUM_ASSETS] = new_tickers1
ind2[:NUM_ASSETS] = new_tickers2
# Crossover weights using arithmetic crossover
new_weights1 = (np.array(weights1) + np.array(weights2)) / 2
new_weights2 = (np.array(weights1) + np.array(weights2)) / 2
# Enforce weight constraints
try:
new_weights1 = generate_weights(NUM_ASSETS, LOWER_WEIGHT, UPPER_WEIGHT)
except ValueError:
new_weights1 = allocate_weights()
try:
new_weights2 = generate_weights(NUM_ASSETS, LOWER_WEIGHT, UPPER_WEIGHT)
except ValueError:
new_weights2 = allocate_weights()
ind1[NUM_ASSETS:] = new_weights1.tolist()
ind2[NUM_ASSETS:] = new_weights2.tolist()
return ind1, ind2
# Register the custom crossover
toolbox.register("mate", cxTickersWeights)
# Genetic operators
toolbox.register("evaluate", evaluate_portfolio)
# Custom mutation: either swap a ticker or adjust weights
def mutate_portfolio(individual, indpb=0.2):
mutation_type = random.choice(['ticker', 'weight'])
if mutation_type == 'ticker':
# Mutation: swap one ticker
idx = random.randint(0, NUM_ASSETS - 1)
current_ticker = individual[idx]
available_tickers = list(set(successful_tickers) - set(individual[:NUM_ASSETS]))
if available_tickers:
new_ticker = random.choice(available_tickers)
individual[idx] = new_ticker
else:
# Mutation: adjust weights
try:
new_weights = generate_weights(NUM_ASSETS, LOWER_WEIGHT, UPPER_WEIGHT)
individual[NUM_ASSETS:] = new_weights.tolist()
except ValueError:
# If unable to generate, leave weights unchanged
pass
return (individual,)
toolbox.register("mutate", mutate_portfolio, indpb=0.2)
toolbox.register("select", tools.selTournament, tournsize=3)
# Determine the number of processes to use
num_processes = max(1, multiprocessing.cpu_count() - 1) # Leave one core free
# Set up multiprocessing pool
pool = multiprocessing.Pool(processes=num_processes)
toolbox.register("map", pool.map)
# Initialize population
print("Initializing population...")
population = toolbox.population(n=POPULATION_SIZE)
# Initialize statistics to keep track
stats = tools.Statistics(lambda ind: ind.fitness.values)
stats.register("avg", np.mean)
stats.register("std", np.std)
stats.register("min", np.min)
stats.register("max", np.max)
# Initialize Hall of Fame to store the best individual
hof = tools.HallOfFame(HALL_OF_FAME_SIZE)
# Define a callback to display better portfolios instantly
class PrintBestPortfolio:
def __init__(self, hof):
self.hof = hof
self.best_fitness = -np.inf
def __call__(self, gen, population, fitnesses):
current_best = max(fitnesses)
if current_best > self.best_fitness:
self.best_fitness = current_best
best_ind = tools.selBest(population, 1)[0]
selected = best_ind[:NUM_ASSETS]
weights = np.array(best_ind[NUM_ASSETS:])
# Training Metrics
portfolio_return_train = np.dot(weights, train_ann_return[selected])
portfolio_volatility_train = np.sqrt(np.dot(weights, np.dot(train_cov_matrix.loc[selected, selected].values, weights)))
sharpe_ratio_train = (portfolio_return_train - RISK_FREE_RATE) / portfolio_volatility_train if portfolio_volatility_train != 0 else 0
portfolio_sortino_train = sortino_ratio(train_returns[selected].dot(weights))
portfolio_cum_returns_train = (1 + train_returns[selected].dot(weights)).cumprod()
portfolio_max_drawdown_train = max_drawdown(portfolio_cum_returns_train)
# Validation Metrics
try:
portfolio_return_val = np.dot(weights, validation_ann_return[selected])
portfolio_volatility_val = np.sqrt(np.dot(weights, np.dot(validation_cov_matrix.loc[selected, selected].values, weights)))
except KeyError as e:
portfolio_return_val = 0
portfolio_volatility_val = 0
sharpe_ratio_val = (portfolio_return_val - RISK_FREE_RATE) / portfolio_volatility_val if portfolio_volatility_val != 0 else 0
portfolio_sortino_val = sortino_ratio(validation_returns[selected].dot(weights))
portfolio_cum_returns_val = (1 + validation_returns[selected].dot(weights)).cumprod()
portfolio_max_drawdown_val = max_drawdown(portfolio_cum_returns_val)
# Display the new best portfolio
portfolio_df = pd.DataFrame({
'Ticker': selected,
'Allocation': [f"{w*100:.2f}%" for w in weights]
})
print(f"\nGeneration {gen}: New Best Portfolio Found!")
display(portfolio_df)
print(f"--- Training Set Performance ---")
print(f"Annualized Return: {portfolio_return_train*100:.2f}%")
print(f"Annualized Volatility: {portfolio_volatility_train*100:.2f}%")
print(f"Sharpe Ratio: {sharpe_ratio_train:.2f}")
print(f"Sortino Ratio: {portfolio_sortino_train:.2f}")
print(f"Maximum Drawdown: {portfolio_max_drawdown_train:.2%}")
print(f"--- Validation Set Performance ---")
print(f"Annualized Return: {portfolio_return_val*100:.2f}%")
print(f"Annualized Volatility: {portfolio_volatility_val*100:.2f}%")
print(f"Sharpe Ratio: {sharpe_ratio_val:.2f}")
print(f"Sortino Ratio: {portfolio_sortino_val:.2f}")
print(f"Maximum Drawdown: {portfolio_max_drawdown_val:.2%}")
# Instantiate the callback
callback = PrintBestPortfolio(hof)
# Begin the evolution
print("\nStarting Genetic Algorithm Evolution...\n")
for gen in tqdm(range(1, MAX_GENERATIONS + 1), desc="Generations"):
# Select the next generation individuals
offspring = toolbox.select(population, len(population))
offspring = list(map(toolbox.clone, offspring))
# Apply crossover on the offspring
for child1, child2 in zip(offspring[::2], offspring[1::2]):
if random.random() < P_CROSSOVER:
toolbox.mate(child1, child2)
del child1.fitness.values
del child2.fitness.values
# Apply mutation on the offspring
for mutant in offspring:
if random.random() < P_MUTATION:
toolbox.mutate(mutant)
del mutant.fitness.values
# Evaluate the individuals with an invalid fitness
invalid_ind = [ind for ind in offspring if not ind.fitness.valid]
fitnesses = toolbox.map(toolbox.evaluate, invalid_ind)
for ind, fit in zip(invalid_ind, fitnesses):
ind.fitness.values = fit
# Replace population with offspring
population[:] = offspring
# Update the hall of fame with the generated individuals
hof.update(population)
# Gather all the fitnesses in one list
fits = [ind.fitness.values[0] for ind in population]
# Callback to print instant best portfolio
callback(gen, population, fits)
print("\nEvolution Complete!\n")
# Retrieve and display the best portfolio found
if len(hof) > 0:
best_ind = hof[0]
selected = best_ind[:NUM_ASSETS]
weights = np.array(best_ind[NUM_ASSETS:])
# Training Metrics
portfolio_return_train = np.dot(weights, train_ann_return[selected])
portfolio_volatility_train = np.sqrt(np.dot(weights, np.dot(train_cov_matrix.loc[selected, selected].values, weights)))
sharpe_ratio_train = (portfolio_return_train - RISK_FREE_RATE) / portfolio_volatility_train if portfolio_volatility_train != 0 else 0
portfolio_sortino_train = sortino_ratio(train_returns[selected].dot(weights))
portfolio_cum_returns_train = (1 + train_returns[selected].dot(weights)).cumprod()
portfolio_max_drawdown_train = max_drawdown(portfolio_cum_returns_train)
# Validation Metrics
try:
portfolio_return_val = np.dot(weights, validation_ann_return[selected])
portfolio_volatility_val = np.sqrt(np.dot(weights, np.dot(validation_cov_matrix.loc[selected, selected].values, weights)))
except KeyError as e:
portfolio_return_val = 0
portfolio_volatility_val = 0
sharpe_ratio_val = (portfolio_return_val - RISK_FREE_RATE) / portfolio_volatility_val if portfolio_volatility_val != 0 else 0
portfolio_sortino_val = sortino_ratio(validation_returns[selected].dot(weights))
portfolio_cum_returns_val = (1 + validation_returns[selected].dot(weights)).cumprod()
portfolio_max_drawdown_val = max_drawdown(portfolio_cum_returns_val)
# Testing Metrics
portfolio_return_test = np.dot(weights, test_ann_return[selected])
portfolio_volatility_test = np.sqrt(np.dot(weights, np.dot(test_cov_matrix.loc[selected, selected].values, weights)))
sharpe_ratio_test = (portfolio_return_test - RISK_FREE_RATE) / portfolio_volatility_test if portfolio_volatility_test != 0 else 0
portfolio_sortino_test = sortino_ratio(test_returns[selected].dot(weights))
portfolio_cum_returns_test = (1 + test_returns[selected].dot(weights)).cumprod()
portfolio_max_drawdown_test = max_drawdown(portfolio_cum_returns_test)
# Composite Fitness Score on Training and Validation Sets
fitness_train = (WEIGHT_SHARPE * sharpe_ratio_train) \
- (WEIGHT_MAX_DRAWDOWN * portfolio_max_drawdown_train) \
+ (WEIGHT_ANNUAL_RETURN * portfolio_return_train) \
- (WEIGHT_VOLATILITY * portfolio_volatility_train) \
+ (WEIGHT_SORTINO * portfolio_sortino_train)
fitness_val = (WEIGHT_SHARPE * sharpe_ratio_val) \
- (WEIGHT_MAX_DRAWDOWN * portfolio_max_drawdown_val) \
+ (WEIGHT_ANNUAL_RETURN * portfolio_return_val) \
- (WEIGHT_VOLATILITY * portfolio_volatility_val) \
+ (WEIGHT_SORTINO * portfolio_sortino_val)
fitness = (fitness_train + fitness_val) / 2
portfolio_df = pd.DataFrame({
'Ticker': selected,
'Allocation': [f"{w*100:.2f}%" for w in weights]
})
print("Optimal Portfolio Found:")
display(portfolio_df)
print(f"--- Training Set Performance ---")
print(f"Annualized Return: {portfolio_return_train*100:.2f}%")
print(f"Annualized Volatility: {portfolio_volatility_train*100:.2f}%")
print(f"Sharpe Ratio: {sharpe_ratio_train:.2f}")
print(f"Sortino Ratio: {portfolio_sortino_train:.2f}")
print(f"Maximum Drawdown: {portfolio_max_drawdown_train:.2%}")
print(f"--- Validation Set Performance ---")
print(f"Annualized Return: {portfolio_return_val*100:.2f}%")
print(f"Annualized Volatility: {portfolio_volatility_val*100:.2f}%")
print(f"Sharpe Ratio: {sharpe_ratio_val:.2f}")
print(f"Sortino Ratio: {portfolio_sortino_val:.2f}")
print(f"Maximum Drawdown: {portfolio_max_drawdown_val:.2%}")
print(f"--- Testing Set Performance ---")
print(f"Annualized Return: {portfolio_return_test*100:.2f}%")
print(f"Annualized Volatility: {portfolio_volatility_test*100:.2f}%")
print(f"Sharpe Ratio: {sharpe_ratio_test:.2f}")
print(f"Sortino Ratio: {portfolio_sortino_test:.2f}")
print(f"Maximum Drawdown: {portfolio_max_drawdown_test:.2%}")
# Assess overfitting
print("\n--- Overfitting Assessment ---")
# Define thresholds for acceptable performance differences
threshold_return = 0.05 # 5% difference in return
threshold_volatility = 0.05 # 5% difference in volatility
train_val_return_diff = portfolio_return_train - portfolio_return_val
train_test_return_diff = portfolio_return_train - portfolio_return_test
train_val_vol_diff = portfolio_volatility_train - portfolio_volatility_val
train_test_vol_diff = portfolio_volatility_train - portfolio_volatility_test
overfitting_return = train_test_return_diff > threshold_return
overfitting_vol = train_test_vol_diff < -threshold_volatility
if overfitting_return and overfitting_vol:
print("Warning: Overfitting detected! The portfolio performs significantly better on training and validation data compared to testing data.")
else:
print("No significant overfitting detected. The portfolio generalizes well to unseen data.")
# Plot cumulative returns for visual assessment
plt.figure(figsize=(14, 7))
# Training Cumulative Returns
plt.subplot(1, 2, 1)
plt.plot(portfolio_cum_returns_train.index, portfolio_cum_returns_train, label='Training')
plt.plot(portfolio_cum_returns_val.index, portfolio_cum_returns_val, label='Validation')
plt.plot(portfolio_cum_returns_test.index, portfolio_cum_returns_test, label='Testing')
plt.title('Cumulative Returns')
plt.xlabel('Date')
plt.ylabel('Cumulative Return')
plt.legend()
# Drawdowns
plt.subplot(1, 2, 2)
plt.plot(portfolio_cum_returns_train.index, (portfolio_cum_returns_train / portfolio_cum_returns_train.cummax()) - 1, label='Training')
plt.plot(portfolio_cum_returns_val.index, (portfolio_cum_returns_val / portfolio_cum_returns_val.cummax()) - 1, label='Validation')
plt.plot(portfolio_cum_returns_test.index, (portfolio_cum_returns_test / portfolio_cum_returns_test.cummax()) - 1, label='Testing')
plt.title('Drawdowns')
plt.xlabel('Date')
plt.ylabel('Drawdown')
plt.legend()
plt.tight_layout()
plt.show()
# Close the multiprocessing pool
pool.close()
pool.join()ASSISTANT
```python
# Install required packages
!pip install yfinance pandas numpy beautifulsoup4 requests tqdm deap matplotlib
# Import necessary libraries
import yfinance as yf
import pandas as pd
import numpy as np
import random
from bs4 import BeautifulSoup
import requests
from tqdm import tqdm
import sys
import warnings
import multiprocessing
import matplotlib.pyplot as plt
from deap import base, creator, tools, algorithms
# Suppress warnings for cleaner output
warnings.filterwarnings("ignore")
# Set random seed for reproducibility
RANDOM_SEED = 42
random.seed(RANDOM_SEED)
np.random.seed(RANDOM_SEED)
# Function to fetch S&P 500 tickers from Wikipedia
def get_sp500_tickers():
url = 'https://en.wikipedia.org/wiki/List_of_S%26P_500_companies'
response = requests.get(url)
if response.status_code != 200:
raise Exception("Failed to fetch S&P 500 tickers from Wikipedia.")
soup = BeautifulSoup(response.text, "lxml")
table = soup.find('table', {'id': 'constituents'})
tickers = []
for row in table.findAll('tr')[1:]:
ticker = row.findAll('td')[0].text.strip()
# Replace '.' with '-' for tickers like BRK.B
ticker = ticker.replace('.', '-')
tickers.append(ticker)
return tickers
# Function to calculate Maximum Drawdown
def max_drawdown(series):
try:
roll_max = series.cummax()
drawdown = (series - roll_max) / roll_max
return drawdown.min()
except:
return -0.0
# Function to calculate Sortino Ratio
def sortino_ratio(returns, target=0):
downside = returns[returns < target]
if len(downside) == 0:
return 0 # No downside risk; assign a neutral value
expected_return = returns.mean() * 252
downside_risk = np.sqrt((downside ** 2).mean()) * np.sqrt(252)
return (expected_return - target) / downside_risk if downside_risk != 0 else 0
# Function to calculate Conditional Value at Risk (CVaR)
def conditional_value_at_risk(returns, confidence_level=0.95):
if len(returns) == 0:
return 0
var = np.percentile(returns, (1 - confidence_level) * 100)
cvar = returns[returns <= var].mean()
return cvar
# Function to generate portfolio weights within constraints
def generate_weights(n, lower=0.10, upper=0.20, max_attempts=1000):
"""
Generates a list of weights that sum to 1.0 with each weight between lower and upper bounds.
"""
for _ in range(max_attempts):
weights = np.random.uniform(lower, upper, n)
weights /= weights.sum()
if all(lower <= w <= upper for w in weights):
return weights
raise ValueError("Unable to generate weights within constraints after multiple attempts.")
# Function to label economic cycles
def label_economic_cycles(dates):
recession_periods = [
('2001-03-01', '2001-11-30'),
('2007-12-01', '2009-06-30'),
('2020-02-01', '2020-04-30')
]
labels = []
for date in dates:
label = 'Growth'
for start, end in recession_periods:
if pd.to_datetime(start) <= date <= pd.to_datetime(end):
label = 'Recession'
break
labels.append(label)
return labels
# Parameters
START_DATE = '2000-01-01'
END_DATE = '2024-10-30'
NUM_ASSETS = 8 # Number of stocks in the portfolio
LOWER_WEIGHT = 0.10
UPPER_WEIGHT = 0.20
RISK_FREE_RATE = 0.017 # Approximate risk-free rate (e.g., 10-year US Treasury yield)
# Step 1: Get S&P 500 tickers
print("Fetching S&P 500 tickers...")
tickers = get_sp500_tickers()
print(f"Number of tickers fetched: {len(tickers)}")
# Step 2: Download historical data with dynamic start date
print("Downloading historical data...")
failed_tickers = []
successful_tickers = []
data = pd.DataFrame()
ticker_first_dates = {}
for ticker in tqdm(tickers, desc="Downloading tickers"):
try:
df = yf.download(ticker, start=START_DATE, end=END_DATE, progress=False)['Adj Close']
if df.empty:
failed_tickers.append(ticker)
else:
data[ticker] = df
successful_tickers.append(ticker)
ticker_first_dates[ticker] = df.first_valid_index()
except Exception as e:
failed_tickers.append(ticker)
print(f"\nTotal tickers with successful data download: {len(successful_tickers)}")
print(f"Total tickers failed to download: {len(failed_tickers)}")
if failed_tickers:
print(f"Failed tickers: {failed_tickers}")
# Determine the earliest start date among all successful tickers
if ticker_first_dates:
earliest_date = min(ticker_first_dates.values())
print(f"Earliest available data starts from: {earliest_date.date()}")
# Adjust the dataset to start from the earliest_date
data = data.loc[earliest_date:]
# Forward fill to propagate last valid observation
data.fillna(method='ffill', inplace=True)
# Backward fill to handle any remaining NaNs at the start
data.fillna(method='bfill', inplace=True)
print(f"Data shape after adjusting start date and dropping incomplete data: {data.shape}")
else:
print("No data available after processing tickers.")
sys.exit()
# If there are not enough tickers, exit
if len(data.columns) < NUM_ASSETS:
print("Not enough tickers with complete data to form a portfolio.")
sys.exit()
# Step 3: Calculate daily returns
returns = data.pct_change().dropna()
# Step 4: Label economic cycles
print("Labeling economic cycles...")
return_dates = returns.index
labels = label_economic_cycles(return_dates)
returns['Economic_Cycle'] = labels
# Step 5: Split data into Training (70%), Validation (15%), Testing (15%)
split_date_1 = int(len(returns) * 0.70)
split_date_2 = int(len(returns) * 0.85)
train_returns = returns.iloc[:split_date_1].drop(columns=['Economic_Cycle'])
train_labels = labels[:split_date_1]
validation_returns = returns.iloc[split_date_1:split_date_2].drop(columns=['Economic_Cycle'])
validation_labels = labels[split_date_1:split_date_2]
test_returns = returns.iloc[split_date_2:].drop(columns=['Economic_Cycle'])
test_labels = labels[split_date_2:]
print(f"Training set: {train_returns.shape[0]} days")
print(f"Validation set: {validation_returns.shape[0]} days")
print(f"Testing set: {test_returns.shape[0]} days")
# Calculate compounded multi-year metrics for each set
def calculate_metrics_multi_year(returns_set, labels_set):
metrics = {}
# Overall
cumulative_return = (1 + returns_set).prod() - 1
metrics['cumulative_return'] = cumulative_return
# Volatility
volatility = returns_set.std() * np.sqrt(252)
metrics['volatility'] = volatility
# Sharpe Ratio
sharpe_ratio = (returns_set.mean() * 252 - RISK_FREE_RATE) / (returns_set.std() * np.sqrt(252))
metrics['sharpe_ratio'] = sharpe_ratio
# Sortino Ratio
sortino = {}
for cycle in ['Growth', 'Recession']:
cycle_returns = returns_set[labels_set == cycle].dropna()
sortino_cycle = sortino_ratio(cycle_returns)
sortino[cycle] = sortino_cycle
metrics['sortino'] = sortino
# CVaR
cvar = {}
for cycle in ['Growth', 'Recession']:
cycle_returns = returns_set[labels_set == cycle].dropna()
cvar_cycle = conditional_value_at_risk(cycle_returns, confidence_level=0.95)
cvar[cycle] = cvar_cycle
metrics['cvar'] = cvar
# Maximum Drawdown
mdd = {}
for cycle in ['Growth', 'Recession']:
cycle_returns = returns_set[labels_set == cycle].dropna()
portfolio_cum_returns = (1 + cycle_returns).cumprod()
mdd_cycle = max_drawdown(portfolio_cum_returns)
mdd[cycle] = mdd_cycle
metrics['max_drawdown'] = mdd
return metrics
train_metrics = calculate_metrics_multi_year(train_returns, train_labels)
validation_metrics = calculate_metrics_multi_year(validation_returns, validation_labels)
test_metrics = calculate_metrics_multi_year(test_returns, test_labels)
# Define Genetic Algorithm parameters
POPULATION_SIZE = 500 # Adjusted for memory constraints
P_CROSSOVER = 0.8 # Probability for crossover
P_MUTATION = 0.2 # Probability for mutating an individual
MAX_GENERATIONS = 100 # Increased number of generations for better convergence
HALL_OF_FAME_SIZE = 1 # Number of best individuals to keep
# Define the weights for each optimization objective
WEIGHT_SHARPE = 0.20
WEIGHT_MAX_DRAWDOWN = 0.15
WEIGHT_CUMULATIVE_RETURN = 0.25
WEIGHT_VOLATILITY = 0.10
WEIGHT_SORTINO_GROWTH = 0.10
WEIGHT_SORTINO_RECESSION = 0.05
WEIGHT_CVAR_GROWTH = 0.10
WEIGHT_CVAR_RECESSION = 0.05
# Define the evaluation (fitness) function
def evaluate_portfolio(individual):
selected_tickers = individual[:NUM_ASSETS]
weights = individual[NUM_ASSETS:]
# Ensure all selected tickers are unique
if len(set(selected_tickers)) != NUM_ASSETS:
return -np.inf, # Invalid individual
# Ensure weights sum to 1 and are within constraints
weights = np.array(weights)
if not np.isclose(weights.sum(), 1.0):
return -np.inf,
if not all(LOWER_WEIGHT <= w <= UPPER_WEIGHT for w in weights):
return -np.inf,
# Check if all tickers are present in the covariance matrix
if not all(ticker in returns.columns for ticker in selected_tickers):
return -np.inf, # Invalid individual
# Calculate portfolio metrics using Training Set
try:
portfolio_returns_train = train_returns[selected_tickers].dot(weights)
portfolio_cum_return_train = (1 + portfolio_returns_train).prod() - 1
portfolio_volatility_train = portfolio_returns_train.std() * np.sqrt(252)
sharpe_ratio_train = (portfolio_returns_train.mean() * 252 - RISK_FREE_RATE) / portfolio_volatility_train if portfolio_volatility_train != 0 else 0
sortino_growth_train = sortino_ratio(train_returns[selected_tickers].dot(weights)[train_labels == 'Growth'])
sortino_recession_train = sortino_ratio(train_returns[selected_tickers].dot(weights)[train_labels == 'Recession'])
cvar_growth_train = conditional_value_at_risk(train_returns[selected_tickers].dot(weights)[train_labels == 'Growth'], 0.95)
cvar_recession_train = conditional_value_at_risk(train_returns[selected_tickers].dot(weights)[train_labels == 'Recession'], 0.95)
portfolio_cum_returns_growth_train = (1 + train_returns[selected_tickers].dot(weights)[train_labels == 'Growth']).cumprod()
portfolio_cum_returns_recession_train = (1 + train_returns[selected_tickers].dot(weights)[train_labels == 'Recession']).cumprod()
portfolio_max_drawdown_growth_train = max_drawdown(portfolio_cum_returns_growth_train)
portfolio_max_drawdown_recession_train = max_drawdown(portfolio_cum_returns_recession_train)
except:
return -np.inf,
# Calculate Validation Metrics
try:
portfolio_returns_val = validation_returns[selected_tickers].dot(weights)
portfolio_cum_return_val = (1 + portfolio_returns_val).prod() - 1
portfolio_volatility_val = portfolio_returns_val.std() * np.sqrt(252)
sharpe_ratio_val = (portfolio_returns_val.mean() * 252 - RISK_FREE_RATE) / portfolio_volatility_val if portfolio_volatility_val != 0 else 0
sortino_growth_val = sortino_ratio(validation_returns[selected_tickers].dot(weights)[validation_labels == 'Growth'])
sortino_recession_val = sortino_ratio(validation_returns[selected_tickers].dot(weights)[validation_labels == 'Recession'])
cvar_growth_val = conditional_value_at_risk(validation_returns[selected_tickers].dot(weights)[validation_labels == 'Growth'], 0.95)
cvar_recession_val = conditional_value_at_risk(validation_returns[selected_tickers].dot(weights)[validation_labels == 'Recession'], 0.95)
portfolio_cum_returns_growth_val = (1 + validation_returns[selected_tickers].dot(weights)[validation_labels == 'Growth']).cumprod()
portfolio_cum_returns_recession_val = (1 + validation_returns[selected_tickers].dot(weights)[validation_labels == 'Recession']).cumprod()
portfolio_max_drawdown_growth_val = max_drawdown(portfolio_cum_returns_growth_val)
portfolio_max_drawdown_recession_val = max_drawdown(portfolio_cum_returns_recession_val)
except:
return -np.inf,
# Combine the metrics into a single fitness score
fitness = (
WEIGHT_SHARPE * sharpe_ratio_train +
WEIGHT_MAX_DRAWDOWN * (portfolio_max_drawdown_growth_train + portfolio_max_drawdown_recession_train) / 2 +
WEIGHT_CUMULATIVE_RETURN * portfolio_cum_return_train +
WEIGHT_VOLATILITY * (-portfolio_volatility_train) +
WEIGHT_SORTINO_GROWTH * sortino_growth_train +
WEIGHT_SORTINO_RECESSION * sortino_recession_train +
WEIGHT_CVAR_GROWTH * cvar_growth_train +
WEIGHT_CVAR_RECESSION * cvar_recession_train
)
return fitness,
# Create the DEAP framework
creator.create("FitnessMax", base.Fitness, weights=(1.0,)) # Single objective, maximize
creator.create("Individual", list, fitness=creator.FitnessMax)
toolbox = base.Toolbox()
# Attribute generators
# Selection part: choosing 8 unique tickers
def select_tickers():
return random.sample(successful_tickers, NUM_ASSETS)
# Allocation part: generating 8 weights between 10% and 20% that sum to 1
def allocate_weights():
return generate_weights(NUM_ASSETS, LOWER_WEIGHT, UPPER_WEIGHT).tolist()
# Define how each individual is created
def create_individual():
tickers = select_tickers()
weights = allocate_weights()
return creator.Individual(tickers + weights)
toolbox.register("individual", create_individual)
toolbox.register("population", tools.initRepeat, list, toolbox.individual)
# Custom Crossover: Ensures tickers remain unique and weights are within constraints
def cxTickersWeights(ind1, ind2):
# Separate tickers and weights
tickers1, weights1 = ind1[:NUM_ASSETS], ind1[NUM_ASSETS:]
tickers2, weights2 = ind2[:NUM_ASSETS], ind2[NUM_ASSETS:]
# Crossover tickers using one-point crossover
cx_point = random.randint(1, NUM_ASSETS -1)
new_tickers1 = tickers1[:cx_point] + [ticker for ticker in tickers2[cx_point:] if ticker not in tickers1[:cx_point]]
new_tickers2 = tickers2[:cx_point] + [ticker for ticker in tickers1[cx_point:] if ticker not in tickers2[:cx_point]]
# Fill the remaining tickers to maintain NUM_ASSETS
available_tickers1 = list(set(successful_tickers) - set(new_tickers1))
available_tickers2 = list(set(successful_tickers) - set(new_tickers2))
while len(new_tickers1) < NUM_ASSETS:
new_tickers1.append(random.choice(available_tickers1))
available_tickers1.remove(new_tickers1[-1])
while len(new_tickers2) < NUM_ASSETS:
new_tickers2.append(random.choice(available_tickers2))
available_tickers2.remove(new_tickers2[-1])
# Assign the new tickers
ind1[:NUM_ASSETS] = new_tickers1
ind2[:NUM_ASSETS] = new_tickers2
# Crossover weights using arithmetic crossover
new_weights1 = (np.array(weights1) + np.array(weights2)) / 2
new_weights2 = (np.array(weights1) + np.array(weights2)) / 2
# Enforce weight constraints
try:
new_weights1 = generate_weights(NUM_ASSETS, LOWER_WEIGHT, UPPER_WEIGHT)
except ValueError:
new_weights1 = allocate_weights()
try:
new_weights2 = generate_weights(NUM_ASSETS, LOWER_WEIGHT, UPPER_WEIGHT)
except ValueError:
new_weights2 = allocate_weights()
ind1[NUM_ASSETS:] = new_weights1.tolist()
ind2[NUM_ASSETS:] = new_weights2.tolist()
return ind1, ind2
# Register the custom crossover
toolbox.register("mate", cxTickersWeights)
# Genetic operators
toolbox.register("evaluate", evaluate_portfolio)
# Custom mutation: either swap a ticker or adjust weights
def mutate_portfolio(individual, indpb=0.2):
mutation_type = random.choice(['ticker', 'weight'])
if mutation_type == 'ticker':
# Mutation: swap one ticker
idx = random.randint(0, NUM_ASSETS - 1)
current_ticker = individual[idx]
available_tickers = list(set(successful_tickers) - set(individual[:NUM_ASSETS]))
if available_tickers:
new_ticker = random.choice(available_tickers)
individual[idx] = new_ticker
else:
# Mutation: adjust weights
try:
new_weights = generate_weights(NUM_ASSETS, LOWER_WEIGHT, UPPER_WEIGHT)
individual[NUM_ASSETS:] = new_weights.tolist()
except ValueError:
# If unable to generate, leave weights unchanged
pass
return (individual,)
toolbox.register("mutate", mutate_portfolio, indpb=0.2)
toolbox.register("select", tools.selTournament, tournsize=3)
# Determine the number of processes to use
num_processes = max(1, multiprocessing.cpu_count() - 1) # Leave one core free
# Set up multiprocessing pool
pool = multiprocessing.Pool(processes=num_processes)
toolbox.register("map", pool.map)
# Initialize population
print("Initializing population...")
population = toolbox.population(n=POPULATION_SIZE)
# Initialize statistics to keep track
stats = tools.Statistics(lambda ind: ind.fitness.values)
stats.register("avg", np.mean)
stats.register("std", np.std)
stats.register("min", np.min)
stats.register("max", np.max)
# Initialize Hall of Fame to store the best individual
hof = tools.HallOfFame(HALL_OF_FAME_SIZE)
# Define a callback to display better portfolios instantly
class PrintBestPortfolio:
def __init__(self, hof):
self.hof = hof
self.best_fitness = -np.inf
def __call__(self, gen, population, fitnesses):
current_best = max(fitnesses)
if current_best > self.best_fitness:
self.best_fitness = current_best
best_ind = tools.selBest(population, 1)[0]
selected = best_ind[:NUM_ASSETS]
weights = np.array(best_ind[NUM_ASSETS:])
# Training Metrics
portfolio_return_train = train_metrics['cumulative_return'][selected]
portfolio_volatility_train = (train_returns[selected].dot(weights)).std() * np.sqrt(252)
sharpe_ratio_train = (train_returns[selected].dot(weights).mean() * 252 - RISK_FREE_RATE) / portfolio_volatility_train if portfolio_volatility_train != 0 else 0
sortino_growth_train = sortino_ratio(train_returns[selected].dot(weights)[train_labels == 'Growth'])
sortino_recession_train = sortino_ratio(train_returns[selected].dot(weights)[train_labels == 'Recession'])
cvar_growth_train = conditional_value_at_risk(train_returns[selected].dot(weights)[train_labels == 'Growth'], 0.95)
cvar_recession_train = conditional_value_at_risk(train_returns[selected].dot(weights)[train_labels == 'Recession'], 0.95)
portfolio_cum_returns_growth_train = (1 + train_returns[selected].dot(weights)[train_labels == 'Growth']).cumprod()
portfolio_cum_returns_recession_train = (1 + train_returns[selected].dot(weights)[train_labels == 'Recession']).cumprod()
portfolio_max_drawdown_growth_train = max_drawdown(portfolio_cum_returns_growth_train)
portfolio_max_drawdown_recession_train = max_drawdown(portfolio_cum_returns_recession_train)
# Validation Metrics
try:
portfolio_return_val = validation_metrics['cumulative_return'][selected]
portfolio_volatility_val = (validation_returns[selected].dot(weights)).std() * np.sqrt(252)
sharpe_ratio_val = (validation_returns[selected].dot(weights).mean() * 252 - RISK_FREE_RATE) / portfolio_volatility_val if portfolio_volatility_val != 0 else 0
sortino_growth_val = sortino_ratio(validation_returns[selected].dot(weights)[validation_labels == 'Growth'])
sortino_recession_val = sortino_ratio(validation_returns[selected].dot(weights)[validation_labels == 'Recession'])
cvar_growth_val = conditional_value_at_risk(validation_returns[selected].dot(weights)[validation_labels == 'Growth'], 0.95)
cvar_recession_val = conditional_value_at_risk(validation_returns[selected].dot(weights)[validation_labels == 'Recession'], 0.95)
portfolio_cum_returns_growth_val = (1 + validation_returns[selected].dot(weights)[validation_labels == 'Growth']).cumprod()
portfolio_cum_returns_recession_val = (1 + validation_returns[selected].dot(weights)[validation_labels == 'Recession']).cumprod()
portfolio_max_drawdown_growth_val = max_drawdown(portfolio_cum_returns_growth_val)
portfolio_max_drawdown_recession_val = max_drawdown(portfolio_cum_returns_recession_val)
except:
portfolio_return_val = 0
portfolio_volatility_val = 0
sharpe_ratio_val = 0
sortino_growth_val = 0
sortino_recession_val = 0
cvar_growth_val = 0
cvar_recession_val = 0
portfolio_max_drawdown_growth_val = 0
portfolio_max_drawdown_recession_val = 0
# Display the new best portfolio
portfolio_df = pd.DataFrame({
'Ticker': selected,
'Allocation': [f"{w*100:.2f}%" for w in weights]
})
print(f"\nGeneration {gen}: New Best Portfolio Found!")
display(portfolio_df)
print(f"--- Training Set Performance ---")
print(f"Cumulative Return: {train_metrics['cumulative_return'][selected]*100:.2f}%")
print(f"Annualized Volatility: {portfolio_volatility_train*100:.2f}%")
print(f"Sharpe Ratio: {sharpe_ratio_train:.2f}")
print(f"Sortino Ratio (Growth): {sortino_growth_train:.2f}")
print(f"Sortino Ratio (Recession): {sortino_recession_train:.2f}")
print(f"CVaR (Growth): {cvar_growth_train:.2%}")
print(f"CVaR (Recession): {cvar_recession_train:.2%}")
print(f"Maximum Drawdown (Growth): {portfolio_max_drawdown_growth_train:.2%}")
print(f"Maximum Drawdown (Recession): {portfolio_max_drawdown_recession_train:.2%}")
print(f"--- Validation Set Performance ---")
print(f"Cumulative Return: {validation_metrics['cumulative_return'][selected]*100:.2f}%")
print(f"Annualized Volatility: {portfolio_volatility_val*100:.2f}%")
print(f"Sharpe Ratio: {sharpe_ratio_val:.2f}")
print(f"Sortino Ratio (Growth): {sortino_growth_val:.2f}")
print(f"Sortino Ratio (Recession): {sortino_recession_val:.2f}")
print(f"CVaR (Growth): {cvar_growth_val:.2%}")
print(f"CVaR (Recession): {cvar_recession_val:.2%}")
print(f"Maximum Drawdown (Growth): {portfolio_max_drawdown_growth_val:.2%}")
print(f"Maximum Drawdown (Recession): {portfolio_max_drawdown_recession_val:.2%}")
# Instantiate the callback
callback = PrintBestPortfolio(hof)
# Begin the evolution
print("\nStarting Genetic Algorithm Evolution...\n")
for gen in tqdm(range(1, MAX_GENERATIONS + 1), desc="Generations"):
# Select the next generation individuals
offspring = toolbox.select(population, len(population))
offspring = list(map(toolbox.clone, offspring))
# Apply crossover on the offspring
for child1, child2 in zip(offspring[::2], offspring[1::2]):
if random.random() < P_CROSSOVER:
toolbox.mate(child1, child2)
del child1.fitness.values
del child2.fitness.values
# Apply mutation on the offspring
for mutant in offspring:
if random.random() < P_MUTATION:
toolbox.mutate(mutant)
del mutant.fitness.values
# Evaluate the individuals with an invalid fitness
invalid_ind = [ind for ind in offspring if not ind.fitness.valid]
fitnesses = toolbox.map(toolbox.evaluate, invalid_ind)
for ind, fit in zip(invalid_ind, fitnesses):
ind.fitness.values = fit
# Replace population with offspring
population[:] = offspring
# Update the hall of fame with the generated individuals
hof.update(population)
# Gather all the fitnesses in one list
fits = [ind.fitness.values[0] for ind in population]
# Callback to print instant best portfolio
callback(gen, population, fits)
print("\nEvolution Complete!\n")
# Retrieve and display the best portfolio found
if len(hof) > 0:
best_ind = hof[0]
selected = best_ind[:NUM_ASSETS]
weights = np.array(best_ind[NUM_ASSETS:])
# Training Metrics
portfolio_returns_train = train_returns[selected].dot(weights)
portfolio_cum_return_train = (1 + portfolio_returns_train).prod() - 1
portfolio_volatility_train = portfolio_returns_train.std() * np.sqrt(252)
sharpe_ratio_train = (portfolio_returns_train.mean() * 252 - RISK_FREE_RATE) / portfolio_volatility_train if portfolio_volatility_train != 0 else 0
sortino_growth_train = sortino_ratio(portfolio_returns_train[train_labels == 'Growth'])
sortino_recession_train = sortino_ratio(portfolio_returns_train[train_labels == 'Recession'])
cvar_growth_train = conditional_value_at_risk(portfolio_returns_train[train_labels == 'Growth'], 0.95)
cvar_recession_train = conditional_value_at_risk(portfolio_returns_train[train_labels == 'Recession'], 0.95)
portfolio_cum_returns_growth_train = (1 + portfolio_returns_train[train_labels == 'Growth']).cumprod()
portfolio_cum_returns_recession_train = (1 + portfolio_returns_train[train_labels == 'Recession']).cumprod()
portfolio_max_drawdown_growth_train = max_drawdown(portfolio_cum_returns_growth_train)
portfolio_max_drawdown_recession_train = max_drawdown(portfolio_cum_returns_recession_train)
# Validation Metrics
try:
portfolio_returns_val = validation_returns[selected].dot(weights)
portfolio_cum_return_val = (1 + portfolio_returns_val).prod() - 1
portfolio_volatility_val = portfolio_returns_val.std() * np.sqrt(252)
sharpe_ratio_val = (portfolio_returns_val.mean() * 252 - RISK_FREE_RATE) / portfolio_volatility_val if portfolio_volatility_val != 0 else 0
sortino_growth_val = sortino_ratio(portfolio_returns_val[validation_labels == 'Growth'])
sortino_recession_val = sortino_ratio(portfolio_returns_val[validation_labels == 'Recession'])
cvar_growth_val = conditional_value_at_risk(portfolio_returns_val[validation_labels == 'Growth'], 0.95)
cvar_recession_val = conditional_value_at_risk(portfolio_returns_val[validation_labels == 'Recession'], 0.95)
portfolio_cum_returns_growth_val = (1 + portfolio_returns_val[validation_labels == 'Growth']).cumprod()
portfolio_cum_returns_recession_val = (1 + portfolio_returns_val[validation_labels == 'Recession']).cumprod()
portfolio_max_drawdown_growth_val = max_drawdown(portfolio_cum_returns_growth_val)
portfolio_max_drawdown_recession_val = max_drawdown(portfolio_cum_returns_recession_val)
except:
portfolio_returns_val = 0
portfolio_volatility_val = 0
sharpe_ratio_val = 0
sortino_growth_val = 0
sortino_recession_val = 0
cvar_growth_val = 0
cvar_recession_val = 0
portfolio_cum_returns_growth_val = 0
portfolio_cum_returns_recession_val = 0
portfolio_max_drawdown_growth_val = 0
portfolio_max_drawdown_recession_val = 0
# Testing Metrics
portfolio_returns_test = test_returns[selected].dot(weights)
portfolio_cum_return_test = (1 + portfolio_returns_test).prod() - 1
portfolio_volatility_test = portfolio_returns_test.std() * np.sqrt(252)
sharpe_ratio_test = (portfolio_returns_test.mean() * 252 - RISK_FREE_RATE) / portfolio_volatility_test if portfolio_volatility_test != 0 else 0
sortino_test_growth = sortino_ratio(portfolio_returns_test[test_labels == 'Growth'])
sortino_test_recession = sortino_ratio(portfolio_returns_test[test_labels == 'Recession'])
cvar_test_growth = conditional_value_at_risk(portfolio_returns_test[test_labels == 'Growth'], 0.95)
cvar_test_recession = conditional_value_at_risk(portfolio_returns_test[test_labels == 'Recession'], 0.95)
portfolio_cum_returns_growth_test = (1 + portfolio_returns_test[test_labels == 'Growth']).cumprod()
portfolio_cum_returns_recession_test = (1 + portfolio_returns_test[test_labels == 'Recession']).cumprod()
portfolio_max_drawdown_growth_test = max_drawdown(portfolio_cum_returns_growth_test)
portfolio_max_drawdown_recession_test = max_drawdown(portfolio_cum_returns_recession_test)
# Composite Fitness Score on Training and Validation Sets
fitness_train = (
WEIGHT_SHARPE * sharpe_ratio_train +
WEIGHT_MAX_DRAWDOWN * (portfolio_max_drawdown_growth_train + portfolio_max_drawdown_recession_train) / 2 +
WEIGHT_CUMULATIVE_RETURN * portfolio_cum_return_train +
WEIGHT_VOLATILITY * (-portfolio_volatility_train) +
WEIGHT_SORTINO_GROWTH * sortino_growth_train +
WEIGHT_SORTINO_RECESSION * sortino_recession_train +
WEIGHT_CVAR_GROWTH * cvar_growth_train +
WEIGHT_CVAR_RECESSION * cvar_recession_train
)
fitness_val = (
WEIGHT_SHARPE * sharpe_ratio_val +
WEIGHT_MAX_DRAWDOWN * (portfolio_max_drawdown_growth_val + portfolio_max_drawdown_recession_val) / 2 +
WEIGHT_CUMULATIVE_RETURN * portfolio_cum_return_val +
WEIGHT_VOLATILITY * (-portfolio_volatility_val) +
WEIGHT_SORTINO_GROWTH * sortino_growth_val +
WEIGHT_SORTINO_RECESSION * sortino_recession_val +
WEIGHT_CVAR_GROWTH * cvar_growth_val +
WEIGHT_CVAR_RECESSION * cvar_recession_val
)
fitness = (fitness_train + fitness_val) / 2
portfolio_df = pd.DataFrame({
'Ticker': selected,
'Allocation': [f"{w*100:.2f}%" for w in weights]
})
print("Optimal Portfolio Found:")
display(portfolio_df)
print(f"--- Training Set Performance ---")
print(f"Cumulative Return: {portfolio_cum_return_train*100:.2f}%")
print(f"Annualized Volatility: {portfolio_volatility_train*100:.2f}%")
print(f"Sharpe Ratio: {sharpe_ratio_train:.2f}")
print(f"Sortino Ratio (Growth): {sortino_growth_train:.2f}")
print(f"Sortino Ratio (Recession): {sortino_recession_train:.2f}")
print(f"CVaR (Growth): {cvar_growth_train:.2%}")
print(f"CVaR (Recession): {cvar_recession_train:.2%}")
print(f"Maximum Drawdown (Growth): {portfolio_max_drawdown_growth_train:.2%}")
print(f"Maximum Drawdown (Recession): {portfolio_max_drawdown_recession_train:.2%}")
print(f"--- Validation Set Performance ---")
print(f"Cumulative Return: {portfolio_cum_return_val*100:.2f}%")
print(f"Annualized Volatility: {portfolio_volatility_val*100:.2f}%")
print(f"Sharpe Ratio: {sharpe_ratio_val:.2f}")
print(f"Sortino Ratio (Growth): {sortino_growth_val:.2f}")
print(f"Sortino Ratio (Recession): {sortino_recession_val:.2f}")
print(f"CVaR (Growth): {cvar_growth_val:.2%}")
print(f"CVaR (Recession): {cvar_recession_val:.2%}")
print(f"Maximum Drawdown (Growth): {portfolio_max_drawdown_growth_val:.2%}")
print(f"Maximum Drawdown (Recession): {portfolio_max_drawdown_recession_val:.2%}")
print(f"--- Testing Set Performance ---")
print(f"Cumulative Return: {portfolio_cum_return_test*100:.2f}%")
print(f"Annualized Volatility: {portfolio_volatility_test*100:.2f}%")
print(f"Sharpe Ratio: {sharpe_ratio_test:.2f}")
print(f"Sortino Ratio (Growth): {sortino_test_growth:.2f}")
print(f"Sortino Ratio (Recession): {sortino_test_recession:.2f}")
print(f"CVaR (Growth): {cvar_test_growth:.2%}")
print(f"CVaR (Recession): {cvar_test_recession:.2%}")
print(f"Maximum Drawdown (Growth): {portfolio_max_drawdown_growth_test:.2%}")
print(f"Maximum Drawdown (Recession): {portfolio_max_drawdown_recession_test:.2%}")
# Assess overfitting
print("\n--- Overfitting Assessment ---")
# Define thresholds for acceptable performance differences
threshold_return = 0.05 # 5% difference in return
threshold_volatility = 0.05 # 5% difference in volatility
train_val_return_diff = portfolio_cum_return_train - portfolio_cum_return_val
train_test_return_diff = portfolio_cum_return_train - portfolio_cum_return_test
train_val_vol_diff = portfolio_volatility_train - portfolio_volatility_val
train_test_vol_diff = portfolio_volatility_train - portfolio_volatility_test
overfitting_return = train_test_return_diff > threshold_return
overfitting_vol = train_test_vol_diff < -threshold_volatility
if overfitting_return and overfitting_vol:
print("Warning: Overfitting detected! The portfolio performs significantly better on training and validation data compared to testing data.")
else:
print("No significant overfitting detected. The portfolio generalizes well to unseen data.")
# Plot cumulative returns for visual assessment
plt.figure(figsize=(14, 7))
# Training Cumulative Returns
plt.subplot(1, 2, 1)
plt.plot(portfolio_cum_returns_growth_train.index, portfolio_cum_returns_growth_train, label='Training Growth')
plt.plot(portfolio_cum_returns_recession_train.index, portfolio_cum_returns_recession_train, label='Training Recession')
plt.plot(portfolio_cum_returns_growth_val.index, portfolio_cum_returns_growth_val, label='Validation Growth')
plt.plot(portfolio_cum_returns_recession_val.index, portfolio_cum_returns_recession_val, label='Validation Recession')
plt.plot(portfolio_cum_returns_growth_test.index, portfolio_cum_returns_growth_test, label='Testing Growth')
plt.plot(portfolio_cum_returns_recession_test.index, portfolio_cum_returns_recession_test, label='Testing Recession')
plt.title('Cumulative Returns')
plt.xlabel('Date')
plt.ylabel('Cumulative Return')
plt.legend()
# Drawdowns
plt.subplot(1, 2, 2)
plt.plot(portfolio_cum_returns_growth_train.index, (portfolio_cum_returns_growth_train / portfolio_cum_returns_growth_train.cummax()) - 1, label='Training Growth Drawdown')
plt.plot(portfolio_cum_returns_recession_train.index, (portfolio_cum_returns_recession_train / portfolio_cum_returns_recession_train.cummax()) - 1, label='Training Recession Drawdown')
plt.plot(portfolio_cum_returns_growth_val.index, (portfolio_cum_returns_growth_val / portfolio_cum_returns_growth_val.cummax()) - 1, label='Validation Growth Drawdown')
plt.plot(portfolio_cum_returns_recession_val.index, (portfolio_cum_returns_recession_val / portfolio_cum_returns_recession_val.cummax()) - 1, label='Validation Recession Drawdown')
plt.plot(portfolio_cum_returns_growth_test.index, (portfolio_cum_returns_growth_test / portfolio_cum_returns_growth_test.cummax()) - 1, label='Testing Growth Drawdown')
plt.plot(portfolio_cum_returns_recession_test.index, (portfolio_cum_returns_recession_test / portfolio_cum_returns_recession_test.cummax()) - 1, label='Testing Recession Drawdown')
plt.title('Drawdowns')
plt.xlabel('Date')
plt.ylabel('Drawdown')
plt.legend()
plt.tight_layout()
plt.show()
# Close the multiprocessing pool
pool.close()
pool.join()
```