"""MULTIPLE EQUITY LINES"""
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import datetime
import matplotlib.dates as mdates
import matplotlib.patches as patches
# --- Funzioni di calcolo delle metriche di trading ---
def calculate_returns(equity_series):
"""Calcola il ritorno percentuale su una serie di equity."""
if equity_series.empty or len(equity_series) < 2:
return 0.0
initial_value = equity_series.iloc[0]
if initial_value == 0:
return np.nan # Ritorno indefinito se si parte da zero
return (equity_series.iloc[-1] / initial_value - 1) * 100
def calculate_max_drawdown(equity_series):
"""Calcola il massimo drawdown percentuale su una serie di equity (versione ottimizzata)."""
if equity_series.empty:
return 0.0
peak = equity_series.expanding().max()
drawdown = (peak - equity_series) / peak
return drawdown.max() * 100
def calculate_sharpe_ratio(equity_series, risk_free_rate=0.0):
"""
Calcola lo Sharpe Ratio.
Non annualizzato, basato su ritorni giornalieri/periodici.
Assumi che risk_free_rate sia già al passo con la frequenza dei dati (es. giornaliero).
"""
if len(equity_series) < 2:
return np.nan
returns = equity_series.pct_change().dropna()
if returns.empty:
return np.nan
std_dev_returns = returns.std()
if std_dev_returns == 0:
return np.nan
excess_returns = returns - risk_free_rate
return excess_returns.mean() / std_dev_returns
def calculate_sortino_ratio(equity_series, risk_free_rate=0.0):
"""
Calcola il Sortino Ratio.
Non annualizzato, basato su ritorni giornalieri/periodici.
Assumi che risk_free_rate sia già al passo con la frequenza dei dati (es. giornaliero).
"""
if len(equity_series) < 2:
return np.nan
returns = equity_series.pct_change().dropna()
if returns.empty:
return np.nan
excess_returns = returns - risk_free_rate
downside_returns = returns[returns < risk_free_rate]
if downside_returns.empty or downside_returns.std() == 0:
return np.nan
downside_deviation = downside_returns.std()
return excess_returns.mean() / downside_deviation
# --- Funzioni helper per la Walk Forward Analysis ---
def _validate_single_equity_line_for_wfo(equity_line: pd.Series, start_date: str, end_date: str):
"""Valida e slice una singola equity line per il periodo WFO."""
if not isinstance(equity_line.index, pd.DatetimeIndex):
raise ValueError("L'indice dell'equity_line deve essere un DatetimeIndex.")
equity_line_sliced = equity_line.loc[start_date:end_date]
if equity_line_sliced.empty:
raise ValueError(f"L'equity line non copre il periodo {start_date} - {end_date} o è vuota.")
if len(equity_line_sliced) < 2:
raise ValueError("L'equity line deve contenere almeno due punti dati per il periodo specificato.")
return equity_line_sliced
def _calculate_wfo_windows(
equity_line_index: pd.DatetimeIndex,
num_cycles: int,
is_percentage: float,
os_percentage: float
) -> list[dict]:
"""
Calcola le finestre di training (IS) e test (OS) per ogni ciclo.
Implementa la logica di scorrimento fornita dall'utente.
Args:
equity_line_index (pd.DatetimeIndex): L'indice di data dell'equity line completa.
num_cycles (int): Numero totale di cicli Walk Forward.
is_percentage (float): Percentuale della durata del ciclo dedicata al periodo In-Sample.
os_percentage (float): Percentuale della durata del ciclo dedicata al periodo Out-of-Sample.
Returns:
Lista di dizionari con chiavi 'cycle', 'train_start_idx', 'train_end_idx',
'test_start_idx', 'test_end_idx' (gli indici di riga originali) e le date corrispondenti.
"""
total_rows = len(equity_line_index)
effective_divisor = ((num_cycles - 1) * os_percentage + 1)
if effective_divisor == 0:
cycle_size_rows = total_rows
else:
cycle_size_rows = int(total_rows // effective_divisor)
oos_size_rows = int(cycle_size_rows * os_percentage)
is_size_rows = cycle_size_rows - oos_size_rows
if is_size_rows < 2: is_size_rows = 2
if oos_size_rows < 2: oos_size_rows = 2
print(f"Durata IS per ciclo (righe): {is_size_rows} giorni")
print(f"Durata OS per ciclo (righe, e passo di avanzamento): {oos_size_rows} giorni\n")
windows_info = []
current_is_start_idx = 0
for i in range(num_cycles):
is_end_idx = current_is_start_idx + is_size_rows - 1
oos_start_idx = is_end_idx + 1
oos_end_idx = oos_start_idx + oos_size_rows - 1
if oos_end_idx >= total_rows or is_end_idx >= total_rows:
print(f"Attenzione: Dati insufficienti per completare il Ciclo {i+1}. Interruzione.")
break
is_start_date = equity_line_index[current_is_start_idx]
is_end_date = equity_line_index[is_end_idx]
oos_start_date = equity_line_index[oos_start_idx]
oos_end_date = equity_line_index[oos_end_idx]
windows_info.append({
'cycle': i + 1,
'train_start_idx': current_is_start_idx,
'train_end_idx': is_end_idx + 1, # +1 per slicing esclusivo
'test_start_idx': oos_start_idx,
'test_end_idx': oos_end_idx + 1, # +1 per slicing esclusivo
'is_start_date': is_start_date,
'is_end_date': is_end_date,
'os_start_date': oos_start_date,
'os_end_date': oos_end_date
})
current_is_start_idx += oos_size_rows
return windows_info
def _calculate_metrics_for_period(equity_series: pd.Series, risk_free_rate: float) -> dict:
"""Calcola un set di metriche per una data serie di equity."""
return {
'Return (%)': calculate_returns(equity_series),
'MDD (%)': calculate_max_drawdown(equity_series),
'Sharpe': calculate_sharpe_ratio(equity_series, risk_free_rate),
'Sortino': calculate_sortino_ratio(equity_series, risk_free_rate),
}
def _process_wfo_cycle(
cycle_num: int,
window_data: dict,
equity_line: pd.Series,
risk_free_rate: float
) -> dict:
"""
Processa un singolo ciclo WFO, calcolando metriche e consistenza.
Le operazioni di calcolo delle metriche interne sono vettorializzate.
"""
is_period_equity = equity_line.iloc[window_data['train_start_idx'] : window_data['train_end_idx']].copy()
os_period_equity = equity_line.iloc[window_data['test_start_idx'] : window_data['test_end_idx']].copy()
if is_period_equity.empty or len(is_period_equity) < 2:
return None # Indica che il ciclo non può essere processato
if os_period_equity.empty or len(os_period_equity) < 2:
return None # Indica che il ciclo non può essere processato
metrics_is = _calculate_metrics_for_period(is_period_equity, risk_free_rate)
metrics_os = _calculate_metrics_for_period(os_period_equity, risk_free_rate)
def calculate_consistency_index(os_stat, is_stat):
if np.isnan(is_stat) or np.isnan(os_stat):
return np.nan
if is_stat == 0:
return np.nan if os_stat == 0 else (np.inf if os_stat > 0 else -np.inf)
return (os_stat - is_stat) / is_stat
cons_ret = calculate_consistency_index(metrics_os['Return (%)'], metrics_is['Return (%)'])
cons_mdd = calculate_consistency_index(metrics_os['MDD (%)'], metrics_is['MDD (%)'])
cons_sharpe = calculate_consistency_index(metrics_os['Sharpe'], metrics_is['Sharpe'])
cons_sortino = calculate_consistency_index(metrics_os['Sortino'], metrics_is['Sortino'])
return {
'Ciclo': f'Ciclo {cycle_num}',
'IS_Start': window_data['is_start_date'].strftime('%Y-%m-%d'),
'IS_End': window_data['is_end_date'].strftime('%Y-%m-%d'),
'OS_Start': window_data['os_start_date'].strftime('%Y-%m-%d'),
'OS_End': window_data['os_end_date'].strftime('%Y-%m-%d'),
'Return_IS (%)': metrics_is['Return (%)'],
'MDD_IS (%)': metrics_is['MDD (%)'],
'Sharpe_IS': metrics_is['Sharpe'],
'Sortino_IS': metrics_is['Sortino'],
'Return_OS (%)': metrics_os['Return (%)'],
'MDD_OS (%)': metrics_os['MDD (%)'],
'Sharpe_OS': metrics_os['Sharpe'],
'Sortino_OS': metrics_os['Sortino'],
'Consistenza_Return': cons_ret,
'Consistenza_MDD': cons_mdd,
'Consistenza_Sharpe': cons_sharpe,
'Consistenza_Sortino': cons_sortino,
}
def _plot_wfo_timeline(equity_line: pd.Series, windows_info: list[dict], os_percentage: float):
"""
Genera il grafico della timeline della Walk Forward Optimization.
Include l'equity line e le barre IS/OS per ogni ciclo.
"""
fig, ax1 = plt.subplots(figsize=(18, 10))
# Plot dell'equity line (opzionale, per contesto visivo)
ax1.plot(equity_line.index, equity_line.values, label='Equity Line di Esempio', color='gray', alpha=0.7, linestyle=':')
ax1.set_ylabel('Valore Equity', color='gray')
ax1.tick_params(axis='y', labelcolor='gray')
# Creazione di un secondo asse y per le barre di timeline
ax2 = ax1.twinx()
train_color = "#2E8B57" # Verde scuro
test_color = "#DC143C" # Rosso
bar_height = 0.6
for i, window_data in enumerate(windows_info):
y_pos = len(windows_info) - i - 1 # Inverti l'ordine per avere il ciclo 1 in alto
train_start_date = window_data['is_start_date']
train_end_date = window_data['is_end_date']
test_start_date = window_data['os_start_date']
test_end_date = window_data['os_end_date']
train_width_days = (train_end_date - train_start_date).days
test_width_days = (test_end_date - test_start_date).days
train_duration_rows = window_data['train_end_idx'] - window_data['train_start_idx']
test_duration_rows = window_data['test_end_idx'] - window_data['test_start_idx']
# Barra di training (verde)
train_rect = patches.Rectangle(
(mdates.date2num(train_start_date), y_pos - bar_height / 2),
train_width_days,
bar_height,
linewidth=1,
edgecolor="black",
facecolor=train_color,
alpha=0.8,
label="Training" if i == 0 else "",
)
ax2.add_patch(train_rect)
# Barra di test (rosso)
test_rect = patches.Rectangle(
(mdates.date2num(test_start_date), y_pos - bar_height / 2),
test_width_days,
bar_height,
linewidth=1,
edgecolor="black",
facecolor=test_color,
alpha=0.8,
label="Test" if i == 0 else "",
)
ax2.add_patch(test_rect)
# Etichetta del ciclo a sinistra
day_width_num = (mdates.date2num(equity_line.index[1]) - mdates.date2num(equity_line.index[0])) if len(equity_line.index) > 1 else 1
ax2.text(
mdates.date2num(train_start_date) - day_width_num * 30,
y_pos,
f"Ciclo {window_data['cycle']}",
ha="right",
va="center",
fontweight="bold",
fontsize=11,
)
# === TESTO SULLE BARRE ===
# Data di inizio training (sopra la barra training, a sinistra)
ax2.text(
mdates.date2num(train_start_date),
y_pos + bar_height / 2 + 0.08,
train_start_date.strftime("%d/%m/%y"),
ha="left",
va="bottom",
fontsize=9,
fontweight="bold",
rotation=0,
color="darkgreen",
)
# Data di fine training / inizio test (una sola data al punto di transizione)
transition_date = train_end_date
ax2.text(
mdates.date2num(transition_date),
y_pos + bar_height / 2 + 0.08,
transition_date.strftime("%d/%m/%y"),
ha="center",
va="bottom",
fontsize=9,
fontweight="bold",
rotation=0,
color="black",
)
# Data di fine test (sopra la barra test, a destra)
ax2.text(
mdates.date2num(test_end_date),
y_pos + bar_height / 2 + 0.08,
test_end_date.strftime("%d/%m/%y"),
ha="right",
va="bottom",
fontsize=9,
fontweight="bold",
rotation=0,
color="darkred",
)
# Durata training (al centro della barra training)
train_center_num = mdates.date2num(train_start_date) + train_width_days / 2
ax2.text(
train_center_num,
y_pos,
f"{train_duration_rows}d",
ha="center",
va="center",
fontsize=11,
fontweight="bold",
color="white",
bbox=dict(
boxstyle="round,pad=0.3",
facecolor="darkgreen",
alpha=0.8,
edgecolor="none",
),
)
# Durata test (al centro della barra test)
test_center_num = mdates.date2num(test_start_date) + test_width_days / 2
ax2.text(
test_center_num,
y_pos,
f"{test_duration_rows}d",
ha="center",
va="center",
fontsize=11,
fontweight="bold",
color="white",
bbox=dict(
boxstyle="round,pad=0.3",
facecolor="darkred",
alpha=0.8,
edgecolor="none",
),
)
# Configurazione degli assi del grafico
ax2.set_xlim(mdates.date2num(equity_line.index.min() - pd.Timedelta(days=50)), mdates.date2num(equity_line.index.max() + pd.Timedelta(days=30)))
ax2.set_ylim(-0.7, len(windows_info) - 0.3)
ax2.xaxis.set_major_formatter(mdates.DateFormatter("%m/%Y"))
plt.setp(ax2.xaxis.get_majorticklabels(), rotation=45)
ax2.set_yticks([]) # Rimuovi i tick dell'asse Y
overall_start_date = windows_info[0]['is_start_date']
overall_end_date = windows_info[-1]['os_end_date']
plt.title(
f"Walk Forward Optimization Timeline\n"
f"{overall_start_date.strftime('%d/%m/%Y')} - {overall_end_date.strftime('%d/%m/%Y')}\n"
f"{len(windows_info)} cicli - {os_percentage * 100:.0f}% test ratio",
fontsize=16,
fontweight="bold",
pad=25,
)
ax1.set_xlabel("Periodo Temporale", fontsize=14) # L'etichetta X va sull'asse principale
# Le legende dei due assi devono essere combinate
lines1, labels1 = ax1.get_legend_handles_labels()
lines2, labels2 = ax2.get_legend_handles_labels()
ax2.legend(lines1 + lines2, labels1 + labels2, loc="upper right", bbox_to_anchor=(1, 1), fontsize=12)
ax2.grid(True, axis="x", alpha=0.3, linestyle="--")
plt.tight_layout()
plt.show()
def _display_results_table(results_df: pd.DataFrame, title: str = "Risultati"):
"""Stampa la tabella dei risultati formattata."""
pd.set_option('display.float_format', lambda x: '%.2f' % x if not pd.isna(x) else 'NaN')
cols_order = ['Ciclo', 'IS_Start', 'IS_End', 'Return_IS (%)', 'MDD_IS (%)', 'Sharpe_IS', 'Sortino_IS',
'OS_Start', 'OS_End', 'Return_OS (%)', 'MDD_OS (%)', 'Sharpe_OS', 'Sortino_OS',
'Consistenza_Return', 'Consistenza_MDD', 'Consistenza_Sharpe', 'Consistenza_Sortino']
# Reindex per garantire l'ordine delle colonne e gestire eventuali colonne mancanti
results_df_display = results_df.reindex(columns=cols_order, fill_value=np.nan)
print(f"\n--- {title} ---")
print(results_df_display.to_string(index=False))
# --- Funzione orchestratrice principale per la Walk Forward Analysis ---
def run_custom_walk_forward_analysis(
start_date: str,
end_date: str,
num_cycles: int,
is_percentage: float,
os_percentage: float,
equity_lines_series: pd.Series, # Ora accetta una Series di equity lines
risk_free_rate: float = 0.0
) -> pd.Series: # Ora restituisce una Series di DataFrames
"""
Orchestra l'esecuzione di un'analisi Walk Forward personalizzata per multiple equity lines.
Args:
start_date (str): Data di inizio del periodo totale (es. '2020-01-01').
end_date (str): Data di fine del periodo totale (es. '2023-12-31').
num_cycles (int): Numero totale di cicli Walk Forward.
is_percentage (float): Percentuale della durata del ciclo dedicata al periodo In-Sample (es. 0.75).
os_percentage (float): Percentuale della durata del ciclo dedicata al periodo Out-of-Sample (es. 0.25).
Deve essere uguale a 1 - is_percentage.
equity_lines_series (pd.Series): Una serie Pandas dove ogni elemento è una pd.Series
rappresentante un'equity line individuale.
risk_free_rate (float): Tasso privo di rischio per Sharpe/Sortino, non annualizzato (es. 0.0 per 0%).
Returns:
pd.Series: Una serie Pandas dove ogni elemento è un pd.DataFrame contenente
i risultati della WFO per la rispettiva equity line.
"""
if not np.isclose(is_percentage + os_percentage, 1.0):
raise ValueError("La somma di is_percentage e os_percentage deve essere 1.0 (o 100%).")
if not (0 < is_percentage < 1) or not (0 < os_percentage < 1):
raise ValueError("is_percentage e os_percentage devono essere tra 0 e 1 (esclusi).")
if num_cycles <= 0:
raise ValueError("Il numero di cicli deve essere maggiore di zero.")
if equity_lines_series.empty:
print("Nessuna equity line fornita per l'analisi.")
return pd.Series(dtype=object)
all_results_dfs = pd.Series(dtype=object) # Per collezionare i DataFrame di tutte le equity lines
try:
# Pre-validazione della prima equity line per calcolare le finestre una sola volta
# Le finestre temporali sono le stesse per tutte le equity line con lo stesso range di date.
print("\n--- Calcolo delle finestre WFO (una tantum) ---")
first_equity_line_for_windows = _validate_single_equity_line_for_wfo(
equity_lines_series.iloc[0], start_date, end_date
)
windows_info = _calculate_wfo_windows(
first_equity_line_for_windows.index, num_cycles, is_percentage, os_percentage
)
if not windows_info:
print("Nessun ciclo Walk Forward valido generato con le impostazioni fornite. Impossibile procedere.")
return pd.Series(dtype=object)
# Generazione del grafico della timeline (uno solo per tutte le analisi)
print("Generazione del grafico della timeline dei periodi WFO...")
_plot_wfo_timeline(first_equity_line_for_windows, windows_info, os_percentage)
print("Grafico della timeline generato.")
print("\n--- Elaborazione dei cicli WFO per ogni equity line ---")
for i, (el_name, current_equity_line) in enumerate(equity_lines_series.items()):
print(f"\n--- Elaborazione Equity Line: '{el_name}' ({i+1}/{len(equity_lines_series)}) ---")
try:
# Validazione e slicing della equity line corrente
current_equity_line_sliced = _validate_single_equity_line_for_wfo(
current_equity_line, start_date, end_date
)
except ValueError as e:
print(f"SALTO: Equity Line '{el_name}' non valida o con dati insufficienti per il periodo: {e}")
all_results_dfs[el_name] = pd.DataFrame() # Inserisce un DataFrame vuoto per quelle saltate
continue
cycle_results_for_current_el = []
for window_data in windows_info: # Riutilizza le stesse finestre calcolate
cycle_result = _process_wfo_cycle(
window_data['cycle'], window_data, current_equity_line_sliced, risk_free_rate
)
if cycle_result:
cycle_results_for_current_el.append(cycle_result)
if cycle_results_for_current_el:
results_df_for_this_el = pd.DataFrame(cycle_results_for_current_el)
all_results_dfs[el_name] = results_df_for_this_el
else:
print(f"Nessun risultato valido generato per l'Equity Line '{el_name}'.")
all_results_dfs[el_name] = pd.DataFrame() # Inserisce un DataFrame vuoto se nessun ciclo valido
print("\n--- Tutte le analisi WFO completate. ---")
return all_results_dfs
except Exception as e:
print(f"Errore critico inaspettato durante l'esecuzione dell'analisi: {e}")
return pd.Series(dtype=object)
# --- Esempio di utilizzo ---
if __name__ == "__main__":
# 1. Dati di input configurabili
start_date = '2020-01-01'
end_date = '2024-12-31' # Circa 5 anni di dati
num_cycles = 10 # Numero desiderato di cicli Walk Forward
is_percentage = 0.75 # Percentuale del ciclo dedicata all'In-Sample
os_percentage = 0.25 # Percentuale del ciclo dedicata all'Out-of-Sample (deve essere 1 - is_percentage)
risk_free_rate_per_period = 0.0000 # Tasso privo di rischio (es. 0% periodico)
# 2. Generazione di 100 Equity Lines casuali di esempio
print("Generazione di 100 equity lines di esempio...")
date_range = pd.date_range(start=start_date, end=end_date, freq='D')
num_days = len(date_range)
initial_equity = 10000
num_equity_lines = 100
all_equity_lines_series = pd.Series(dtype=object) # Una Series che conterrà 100 DataFrame
np.random.seed(42) # Per riproducibilità dei risultati delle equity lines
for j in range(num_equity_lines):
# Genera ritorni giornalieri leggermente diversi per ogni equity line
# Aggiunta di piccole variazioni per rendere le equity line distinte
loc_variation = (j / num_equity_lines) * 0.0001 # Variazione della media dei ritorni
scale_variation = (j / num_equity_lines) * 0.00005 # Variazione della std dev dei ritorni
daily_returns = np.random.normal(loc=0.0005 + loc_variation,
scale=0.005 + scale_variation,
size=num_days)
equity_values = initial_equity * (1 + daily_returns).cumprod()
# Assicurati che l'equity non scenda troppo per avere dati significativi
equity_values[equity_values < 0.1 * initial_equity] = 0.1 * initial_equity
all_equity_lines_series[f'Equity_Line_{j+1}'] = pd.Series(equity_values, index=date_range)
print(f"Generate {len(all_equity_lines_series)} equity lines di esempio.")
# 3. Esegui l'analisi Walk Forward per tutte le equity lines
results_series_of_dfs = run_custom_walk_forward_analysis(
start_date=start_date,
end_date=end_date,
num_cycles=num_cycles,
is_percentage=is_percentage,
os_percentage=os_percentage,
equity_lines_series=all_equity_lines_series, # Passa la Series di equity lines
risk_free_rate=risk_free_rate_per_period
)
# Mostra un riepilogo dei risultati e qualche tabella di esempio
if not results_series_of_dfs.empty:
print(f"\nAnalisi completata. Ottenute {len(results_series_of_dfs)} tabelle di risultati.")
print("\n--- Anteprima delle prime 3 tabelle di risultati ---")
for k, (el_name, df_result) in enumerate(results_series_of_dfs.head(3).items()):
_display_results_table(df_result, title=f"Risultati per {el_name}")
if k < 2: # Aggiungi una riga di separazione tra le tabelle di anteprima
print("\n" + "="*80 + "\n")
else:
print("\nNessun risultato disponibile.")
Details