Ottimizzatore/Ottimizzatore UK.py

import pandas as pd
import numpy as np
from sqlalchemy import create_engine, text
from sqlalchemy.exc import SQLAlchemyError
from pypfopt import EfficientFrontier, risk_models, expected_returns
from pypfopt.exceptions import OptimizationError
import matplotlib.pyplot as plt
import os
import sys
from datetime import datetime

# Cartelle di input/output/plot
OUTPUT_DIR = "Output"
INPUT_DIR = "Input"
PLOT_DIR = "Plot"

os.makedirs(OUTPUT_DIR, exist_ok=True)
os.makedirs(INPUT_DIR, exist_ok=True)
os.makedirs(PLOT_DIR, exist_ok=True)

def excel_path(filename: str) -> str:
    """Percorso completo per i file Excel di output."""
    return os.path.join(OUTPUT_DIR, filename)

def plot_path(filename: str) -> str:
    """Percorso completo per i file di grafico."""
    return os.path.join(PLOT_DIR, filename)

# Configurazione della connessione al database (da connection.txt)
params = {}
with open("connection.txt", "r") as f:
    for line in f:
        line = line.strip()
        if line and not line.startswith('#'):
            key, value = line.split('=', 1)
            params[key.strip()] = value.strip()

username = params.get('username')
password = params.get('password')
host = params.get('host')
port = params.get('port', '1433')
database = params.get('database')
connection_string = (
    f"mssql+pyodbc://{username}:{password}@{host}:{port}/{database}?driver=ODBC+Driver+17+for+SQL+Server"
)

try:
    engine = create_engine(connection_string)
    with engine.connect() as connection:
        _ = connection.execute(text("SELECT 1"))
        print("Connessione al database riuscita.")
except SQLAlchemyError as e:
    print("Errore durante la connessione al database:", e)
    sys.exit()

# Caricamento del template Excel
template_path = os.path.join(INPUT_DIR, 'Template_Guardian.xls')
template_df = pd.read_excel(template_path)

# Caricamento dati degli ISIN
file_path = os.path.join(INPUT_DIR, 'Universo ETF per ottimizzatore UK.xlsx')
df = pd.read_excel(file_path, usecols=['ISIN', 'Nome', 'Categoria', 'Asset Class', 'PesoMax', 'Codice Titolo'],dtype={'Codice Titolo':str})

# Intervallo di date degli ultimi 5 anni, escludendo sabati e domeniche
end_date = pd.Timestamp.now().normalize() - pd.Timedelta(days=1)
start_date = end_date - pd.DateOffset(years=5)
all_dates = pd.date_range(start=start_date, end=end_date, freq='B').normalize()

# DataFrame vuoto con le date come indice
final_df = pd.DataFrame(index=all_dates)

# Iterazione sugli ISIN e recupero dei dati
isin_from_db = set()
for isin in df['ISIN'].unique():
    print(f"Working on ISIN: {isin}")
    procedure_call = f"EXEC opt_RendimentoGiornaliero1_GBP @ISIN = '{isin}', @n = 1305"
    try:
        temp_df = pd.read_sql_query(procedure_call, engine)
        if temp_df.empty:
            print(f"Nessun dato recuperato per {isin}, skipping...")
            continue
        temp_df['Px_Date'] = pd.to_datetime(temp_df['Px_Date'], format='%Y-%m-%d').dt.normalize()
        temp_df.set_index('Px_Date', inplace=True)
        temp_df['RendimentoGiornaliero'] = temp_df['RendimentoGiornaliero'] / 100
        final_df[isin] = temp_df['RendimentoGiornaliero'].reindex(all_dates)
        isin_from_db.add(isin)
        print(f"Dati recuperati per {isin}: {final_df[isin].count()} righe di dati non-null prelevate.")
    except SQLAlchemyError as e:
        print(f"Errore durante l'esecuzione della stored procedure per {isin}:", e)

final_df.fillna(0, inplace=True)

# Configurazione degli obiettivi di volatilità
volatility_targets = {
    (5, 0.06): 'VAR3_GBP',
    #(1, 0.12): 'VAR6_1Y',
    #(3, 0.12): 'VAR6_3Y',
    (5, 0.12): 'VAR6_GBP',
    (5, 0.18): 'VAR9_GBP'
}

# Definizione del numero di giorni lavorativi per anno
days_per_year = 252
riskfree_rate = 0.02

# Ottimizzazione per ciascun target di volatilità e salvataggio dei risultati
optimized_weights = pd.DataFrame()
summary_data = []
export_rows = []
for (years, target_vol), name in volatility_targets.items():
    period_start_date = end_date - pd.DateOffset(years=years)
    period_df = final_df.loc[period_start_date:end_date]
    
    # Calcolo dei parametri per l'ottimizzazione
    daily_returns_mean = period_df.mean()
    annual_returns_mean = daily_returns_mean * days_per_year
    annual_covariance_matrix = risk_models.sample_cov(period_df, returns_data=True)
    
    ef = EfficientFrontier(annual_returns_mean, annual_covariance_matrix)
    
    # Aggiunta dei vincoli per le categorie e le asset class
    categories_limits = df.groupby('Categoria')['PesoMax'].max().to_dict()
    asset_class_limits = {
        'Azionari': 0.75,
        'Obbligazionari': 0.75,
        'Metalli Preziosi': 0.20,
        'Materie Prime': 0.05,
        'Immobiliare': 0.05
    }
    
    for category, max_weight in categories_limits.items():
        isin_list = df[df['Categoria'] == category]['ISIN'].tolist()
        category_idx = [period_df.columns.get_loc(isin) for isin in isin_list if isin in period_df.columns]
        ef.add_constraint(lambda w: sum(w[i] for i in category_idx) <= max_weight)
        ef.add_constraint(lambda w: sum(w[i] for i in category_idx) >= 0)
    
    for asset_class, max_weight in asset_class_limits.items():
        isin_list = df[df['Asset Class'] == asset_class]['ISIN'].tolist()
        asset_class_idx = [period_df.columns.get_loc(isin) for isin in isin_list if isin in period_df.columns]
        ef.add_constraint(lambda w: sum(w[i] for i in asset_class_idx) <= max_weight)
        ef.add_constraint(lambda w: sum(w[i] for i in asset_class_idx) >= 0)

    try:
        ef.efficient_risk(target_volatility=target_vol)
        weights = ef.clean_weights()
        optimized_weights[name] = pd.Series(weights)
        exp_ret, exp_vol, sharpe = ef.portfolio_performance(verbose=True, risk_free_rate=riskfree_rate)
        summary_data.append({
            "Portafoglio": name,
            "Years": years,
            "Target Vol": f"{target_vol:.2%}",
            "Expected annual return": f"{exp_ret:.2%}",
            "Annual volatility": f"{exp_vol:.2%}",
            "Sharpe Ratio": f"{sharpe:.2f}",
        })
                
        # Creazione del DataFrame per i risultati
        results = []
        for isin, weight in weights.items():
            if weight > 0:
                codice_titolo = df.loc[df['ISIN'] == isin, 'Codice Titolo'].values[0]
                nome = df.loc[df['ISIN'] == isin, 'Nome'].values[0]
                results.append({
                    'cod_por': f'PTFOPT{name}',  # Nome del portafoglio
                    'cod_tit': codice_titolo,   # Codice titolo
                    'des_tit': nome,            # Nome titolo
                    'peso': weight * 99         # Peso, moltiplicato per 99
                })

        # Creazione del DataFrame per i risultati
        results_df = pd.DataFrame(results)

        # Accumula le righe per esportazione unica
        export_rows.append(results_df)
        
        # Grafico a torta per ciascun portafoglio ottimizzato
        asset_allocations = {asset: 0 for asset in asset_class_limits}
        for isin, weight in weights.items():
            asset_class = df.loc[df['ISIN'] == isin, 'Asset Class'].values[0]
            asset_allocations[asset_class] += weight
        plt.figure(figsize=(8, 6))
        plt.pie(asset_allocations.values(), labels=asset_allocations.keys(), autopct='%1.1f%%')
        plt.title(f'Asset Allocation for {name}')
        pie_path = plot_path(f'Asset_Allocation_{name}.png')
        plt.savefig(pie_path, dpi=150, bbox_inches='tight')
        plt.close()

    except OptimizationError as e:
        print(f"Optimization failed for {name}: {e}")
        optimized_weights[name] = pd.Series([0] * len(annual_returns_mean))

# Aggiunta del nome dell'ETF nel DataFrame optimized_weights
optimized_weights_with_names = optimized_weights.copy()
optimized_weights_with_names['Nome ETF'] = [df.loc[df['ISIN'] == isin, 'Nome'].values[0] for isin in optimized_weights.index]

summary_df = pd.DataFrame(summary_data)
output_path = excel_path("Riepilogo pesi GBP.xlsx")
with pd.ExcelWriter(output_path, engine='openpyxl', mode='w') as writer:
    optimized_weights_with_names.to_excel(writer, sheet_name='Pesi Ottimizzati', index=True)
    summary_df.to_excel(writer, sheet_name='metriche', index=False)

print(f"All optimized weights saved to '{output_path}' (with metriche).")

# Export unico con tutti i pesi in un solo foglio
date_tag = datetime.now().strftime("%Y%m%d")
combined_path = excel_path(f"{date_tag} Pesi ottimizzati UK.xlsx")
with pd.ExcelWriter(combined_path, engine='openpyxl', mode='w') as writer:
    if export_rows:
        combined_df = pd.concat([template_df] + export_rows, ignore_index=True)
    else:
        combined_df = template_df.copy()
    combined_df.to_excel(writer, sheet_name='Pesi Ottimizzati', index=False)
print(f"Pesi ottimizzati salvati in un unico file/sheet: '{combined_path}'.")