wip refactor
This commit is contained in:
giuseppenuc committed
1 parent
549ced1aea
commit
d092c0967c
74 files changed
+2008
-1095015
No files matched your search
Whitespace-only changes.
Binary file not shown.
Binary file not shown.
+166
-2
@@ -2,7 +2,8 @@ import psutil
|
||||
import multiprocessing
|
||||
import re
|
||||
import pandas as pd
|
||||
from typing import List
|
||||
from typing import List, Dict
|
||||
import numpy as np
|
||||
|
||||
|
||||
def get_optimal_workers() -> int:
|
||||
@@ -212,4 +213,167 @@ def get_full_data(simulated_data: pd.DataFrame,
|
||||
full_data[f'{col}_ma3'] = full_data[col].rolling(window=3, min_periods=1).mean()
|
||||
full_data[f'{col}_ma5'] = full_data[col].rolling(window=5, min_periods=1).mean()
|
||||
|
||||
return full_data
|
||||
return full_data
|
||||
|
||||
|
||||
|
||||
|
||||
import numpy as np
|
||||
from typing import List, Dict
|
||||
|
||||
def prepare_static_features_multiple(varieties_info: List[Dict],
|
||||
percentages: List[float],
|
||||
hectares: float,
|
||||
all_varieties: List[str]) -> np.ndarray:
|
||||
"""
|
||||
Prepara le feature statiche per multiple varietà.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
varieties_info : List[Dict]
|
||||
Lista di dizionari contenenti le informazioni sulle varietà selezionate
|
||||
percentages : List[float]
|
||||
Lista delle percentuali corrispondenti a ciascuna varietà selezionata
|
||||
hectares : float
|
||||
Numero di ettari totali
|
||||
all_varieties : List[str]
|
||||
Lista di tutte le possibili varietà nel dataset originale
|
||||
|
||||
Returns
|
||||
-------
|
||||
np.ndarray
|
||||
Array numpy contenente tutte le feature statiche
|
||||
"""
|
||||
# Inizializza un dizionario per tutte le varietà possibili
|
||||
variety_data = {variety.lower(): {
|
||||
'pct': 0,
|
||||
'prod_t_ha': 0,
|
||||
'tech': '',
|
||||
'oil_prod_t_ha': 0,
|
||||
'oil_prod_l_ha': 0,
|
||||
'min_yield_pct': 0,
|
||||
'max_yield_pct': 0,
|
||||
'min_oil_prod_l_ha': 0,
|
||||
'max_oil_prod_l_ha': 0,
|
||||
'avg_oil_prod_l_ha': 0,
|
||||
'l_per_t': 0,
|
||||
'min_l_per_t': 0,
|
||||
'max_l_per_t': 0,
|
||||
'avg_l_per_t': 0,
|
||||
'water_need_spring': 0,
|
||||
'water_need_summer': 0,
|
||||
'water_need_autumn': 0,
|
||||
'water_need_winter': 0,
|
||||
'annual_water_need': 0,
|
||||
'optimal_temp': 0,
|
||||
'drought_resistance': 0
|
||||
} for variety in all_varieties}
|
||||
|
||||
# Aggiorna i dati per le varietà selezionate
|
||||
for variety_info, percentage in zip(varieties_info, percentages):
|
||||
variety_name = clean_column_name(variety_info['variet_di_olive']).lower()
|
||||
technique = clean_column_name(variety_info['tecnica_di_coltivazione']).lower()
|
||||
|
||||
if variety_name not in variety_data:
|
||||
print(f"Attenzione: La varietà '{variety_name}' non è presente nella lista delle varietà conosciute.")
|
||||
continue
|
||||
|
||||
variety_data[variety_name].update({
|
||||
'pct': percentage / 100,
|
||||
'prod_t_ha': variety_info['produzione_tonnellateettaro'],
|
||||
'tech': technique,
|
||||
'oil_prod_t_ha': variety_info['produzione_olio_tonnellateettaro'],
|
||||
'oil_prod_l_ha': variety_info['produzione_olio_litriettaro'],
|
||||
'min_yield_pct': variety_info['min__resa'],
|
||||
'max_yield_pct': variety_info['max__resa'],
|
||||
'min_oil_prod_l_ha': variety_info['min_produzione_olio_litriettaro'],
|
||||
'max_oil_prod_l_ha': variety_info['max_produzione_olio_litriettaro'],
|
||||
'avg_oil_prod_l_ha': variety_info['media_produzione_olio_litriettaro'],
|
||||
'l_per_t': variety_info['litri_per_tonnellata'],
|
||||
'min_l_per_t': variety_info['min_litri_per_tonnellata'],
|
||||
'max_l_per_t': variety_info['max_litri_per_tonnellata'],
|
||||
'avg_l_per_t': variety_info['media_litri_per_tonnellata'],
|
||||
'water_need_spring': variety_info['fabbisogno_acqua_primavera_mettaro'],
|
||||
'water_need_summer': variety_info['fabbisogno_acqua_estate_mettaro'],
|
||||
'water_need_autumn': variety_info['fabbisogno_acqua_autunno_mettaro'],
|
||||
'water_need_winter': variety_info['fabbisogno_acqua_inverno_mettaro'],
|
||||
'annual_water_need': variety_info['fabbisogno_idrico_annuale_mettaro'],
|
||||
'optimal_temp': variety_info['temperatura_ottimale'],
|
||||
'drought_resistance': variety_info['resistenza_alla_siccit']
|
||||
})
|
||||
|
||||
# Crea il vettore delle feature
|
||||
static_features = [hectares]
|
||||
|
||||
# Lista delle feature per ogni varietà
|
||||
variety_features = ['pct', 'prod_t_ha', 'oil_prod_t_ha', 'oil_prod_l_ha',
|
||||
'min_yield_pct', 'max_yield_pct', 'min_oil_prod_l_ha',
|
||||
'max_oil_prod_l_ha', 'avg_oil_prod_l_ha', 'l_per_t',
|
||||
'min_l_per_t', 'max_l_per_t', 'avg_l_per_t',
|
||||
'water_need_spring', 'water_need_summer', 'water_need_autumn',
|
||||
'water_need_winter', 'annual_water_need', 'optimal_temp',
|
||||
'drought_resistance']
|
||||
|
||||
# Appiattisci i dati delle varietà
|
||||
for variety in all_varieties:
|
||||
variety_lower = variety.lower()
|
||||
# Feature esistenti
|
||||
for feature in variety_features:
|
||||
static_features.append(variety_data[variety_lower][feature])
|
||||
|
||||
# Feature binarie per le tecniche
|
||||
for technique in ['tradizionale', 'intensiva', 'superintensiva']:
|
||||
static_features.append(1 if variety_data[variety_lower]['tech'] == technique else 0)
|
||||
|
||||
return np.array(static_features).reshape(1, -1)
|
||||
|
||||
|
||||
def get_feature_names(all_varieties: List[str]) -> List[str]:
|
||||
"""
|
||||
Genera i nomi delle feature nell'ordine corretto.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
all_varieties : List[str]
|
||||
Lista di tutte le varietà possibili
|
||||
|
||||
Returns
|
||||
-------
|
||||
List[str]
|
||||
Lista dei nomi delle feature
|
||||
"""
|
||||
feature_names = ['hectares']
|
||||
|
||||
variety_features = ['pct', 'prod_t_ha', 'oil_prod_t_ha', 'oil_prod_l_ha',
|
||||
'min_yield_pct', 'max_yield_pct', 'min_oil_prod_l_ha',
|
||||
'max_oil_prod_l_ha', 'avg_oil_prod_l_ha', 'l_per_t',
|
||||
'min_l_per_t', 'max_l_per_t', 'avg_l_per_t']
|
||||
|
||||
techniques = ['tradizionale', 'intensiva', 'superintensiva']
|
||||
|
||||
for variety in all_varieties:
|
||||
for feature in variety_features:
|
||||
feature_names.append(f"{variety}_{feature}")
|
||||
for technique in techniques:
|
||||
feature_names.append(f"{variety}_tech_{technique}")
|
||||
|
||||
return feature_names
|
||||
|
||||
def add_controlled_variation(base_value: float, max_variation_pct: float = 0.20) -> float:
|
||||
"""
|
||||
Aggiunge una variazione controllata a un valore base.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
base_value : float
|
||||
Valore base da modificare
|
||||
max_variation_pct : float
|
||||
Percentuale massima di variazione (default 20%)
|
||||
|
||||
Returns
|
||||
-------
|
||||
float
|
||||
Valore con variazione applicata
|
||||
"""
|
||||
variation = np.random.uniform(-max_variation_pct, max_variation_pct)
|
||||
return base_value * (1 + variation)
|
||||
@@ -1,282 +0,0 @@
|
||||
import numpy as np
|
||||
from typing import Dict, Tuple, List, Optional
|
||||
from scipy import stats
|
||||
|
||||
|
||||
def calculate_real_error(
|
||||
model,
|
||||
test_data: Dict,
|
||||
test_targets: np.ndarray,
|
||||
scaler_y,
|
||||
target_names: Optional[List[str]] = None
|
||||
) -> Tuple[List[float], List[float]]:
|
||||
"""
|
||||
Calcola l'errore reale denormalizzando le predizioni.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
model : tf.keras.Model
|
||||
Modello addestrato
|
||||
test_data : dict
|
||||
Dati di test
|
||||
test_targets : np.ndarray
|
||||
Target di test
|
||||
scaler_y : scaler
|
||||
Scaler utilizzato per normalizzare i target
|
||||
target_names : list, optional
|
||||
Nomi dei target
|
||||
|
||||
Returns
|
||||
-------
|
||||
tuple
|
||||
(percentage_errors, absolute_errors)
|
||||
"""
|
||||
# Predizioni
|
||||
predictions = model.predict(test_data)
|
||||
|
||||
# Denormalizza predizioni e target
|
||||
predictions_real = scaler_y.inverse_transform(predictions)
|
||||
targets_real = scaler_y.inverse_transform(test_targets)
|
||||
|
||||
# Calcola errori percentuali e assoluti
|
||||
percentage_errors = []
|
||||
absolute_errors = []
|
||||
|
||||
if target_names is None:
|
||||
target_names = [f'target_{i}' for i in range(predictions_real.shape[1])]
|
||||
|
||||
# Calcola errori per ogni target
|
||||
for i in range(predictions_real.shape[1]):
|
||||
mae = np.mean(np.abs(predictions_real[:, i] - targets_real[:, i]))
|
||||
mape = np.mean(np.abs((predictions_real[:, i] - targets_real[:, i]) / targets_real[:, i])) * 100
|
||||
percentage_errors.append(mape)
|
||||
absolute_errors.append(mae)
|
||||
|
||||
print(f"\n{target_names[i]}:")
|
||||
print(f"MAE assoluto: {mae:.2f}")
|
||||
print(f"Errore percentuale medio: {mape:.2f}%")
|
||||
print(f"Precisione: {100 - mape:.2f}%")
|
||||
print("-" * 50)
|
||||
|
||||
return percentage_errors, absolute_errors
|
||||
|
||||
|
||||
def evaluate_model_performance(
|
||||
model,
|
||||
data: Dict,
|
||||
targets: np.ndarray,
|
||||
set_name: str = "",
|
||||
threshold: Optional[float] = None
|
||||
) -> Dict:
|
||||
"""
|
||||
Valuta le performance del modello su un set di dati.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
model : tf.keras.Model
|
||||
Modello da valutare
|
||||
data : dict
|
||||
Dati di input
|
||||
targets : np.ndarray
|
||||
Target reali
|
||||
set_name : str
|
||||
Nome del set di dati
|
||||
threshold : float, optional
|
||||
Soglia per calcolare accuracy binaria
|
||||
|
||||
Returns
|
||||
-------
|
||||
dict
|
||||
Dizionario con le metriche calcolate
|
||||
"""
|
||||
predictions = model.predict(data, verbose=0)
|
||||
metrics = {}
|
||||
|
||||
target_names = ['olive_prod', 'min_oil_prod', 'max_oil_prod', 'avg_oil_prod', 'total_water_need']
|
||||
|
||||
for i, name in enumerate(target_names):
|
||||
# Metriche di base
|
||||
mae = np.mean(np.abs(targets[:, i] - predictions[:, i]))
|
||||
mse = np.mean(np.square(targets[:, i] - predictions[:, i]))
|
||||
rmse = np.sqrt(mse)
|
||||
mape = np.mean(np.abs((targets[:, i] - predictions[:, i]) / (targets[:, i] + 1e-7))) * 100
|
||||
|
||||
# R2 score
|
||||
ss_res = np.sum(np.square(targets[:, i] - predictions[:, i]))
|
||||
ss_tot = np.sum(np.square(targets[:, i] - np.mean(targets[:, i])))
|
||||
r2 = 1 - (ss_res / (ss_tot + 1e-7))
|
||||
|
||||
# Salva le metriche
|
||||
metrics[f"{name}_mae"] = mae
|
||||
metrics[f"{name}_rmse"] = rmse
|
||||
metrics[f"{name}_mape"] = mape
|
||||
metrics[f"{name}_r2"] = r2
|
||||
|
||||
# Calcola accuracy binaria se fornita una soglia
|
||||
if threshold is not None:
|
||||
binary_acc = np.mean(
|
||||
(predictions[:, i] > threshold) == (targets[:, i] > threshold)
|
||||
)
|
||||
metrics[f"{name}_binary_acc"] = binary_acc
|
||||
|
||||
if set_name:
|
||||
print(f"\nPerformance sul set {set_name}:")
|
||||
for metric, value in metrics.items():
|
||||
print(f"{metric}: {value:.4f}")
|
||||
|
||||
return metrics
|
||||
|
||||
|
||||
def calculate_efficiency_metrics(
|
||||
predictions: np.ndarray,
|
||||
targets: np.ndarray,
|
||||
resource_usage: np.ndarray
|
||||
) -> Dict:
|
||||
"""
|
||||
Calcola metriche di efficienza basate sull'utilizzo delle risorse.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
predictions : np.ndarray
|
||||
Predizioni del modello
|
||||
targets : np.ndarray
|
||||
Target reali
|
||||
resource_usage : np.ndarray
|
||||
Dati sull'utilizzo delle risorse
|
||||
|
||||
Returns
|
||||
-------
|
||||
dict
|
||||
Metriche di efficienza
|
||||
"""
|
||||
metrics = {}
|
||||
|
||||
# Efficienza di produzione
|
||||
production_efficiency = predictions / (resource_usage + 1e-7)
|
||||
target_efficiency = targets / (resource_usage + 1e-7)
|
||||
|
||||
# Calcola metriche
|
||||
metrics['mean_efficiency'] = np.mean(production_efficiency)
|
||||
metrics['efficiency_error'] = np.mean(np.abs(production_efficiency - target_efficiency))
|
||||
metrics['efficiency_std'] = np.std(production_efficiency)
|
||||
|
||||
# ROI stimato
|
||||
estimated_roi = (predictions - resource_usage) / (resource_usage + 1e-7)
|
||||
actual_roi = (targets - resource_usage) / (resource_usage + 1e-7)
|
||||
metrics['roi_error'] = np.mean(np.abs(estimated_roi - actual_roi))
|
||||
|
||||
# Sostenibilità
|
||||
metrics['resource_utilization'] = np.mean(predictions / resource_usage)
|
||||
metrics['efficiency_improvement'] = (
|
||||
np.mean(production_efficiency) - np.mean(target_efficiency)
|
||||
) / np.mean(target_efficiency) * 100
|
||||
|
||||
return metrics
|
||||
|
||||
|
||||
def calculate_forecast_accuracy(
|
||||
predictions: np.ndarray,
|
||||
targets: np.ndarray,
|
||||
horizons: List[int]
|
||||
) -> Dict:
|
||||
"""
|
||||
Calcola l'accuratezza delle previsioni per diversi orizzonti temporali.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
predictions : np.ndarray
|
||||
Predizioni del modello
|
||||
targets : np.ndarray
|
||||
Target reali
|
||||
horizons : list
|
||||
Lista degli orizzonti temporali da valutare
|
||||
|
||||
Returns
|
||||
-------
|
||||
dict
|
||||
Accuratezza per ogni orizzonte
|
||||
"""
|
||||
accuracy_metrics = {}
|
||||
|
||||
for horizon in horizons:
|
||||
# Seleziona dati per l'orizzonte corrente
|
||||
pred_horizon = predictions[:-horizon]
|
||||
target_horizon = targets[horizon:]
|
||||
|
||||
# Calcola metriche
|
||||
mae = np.mean(np.abs(pred_horizon - target_horizon))
|
||||
mape = np.mean(np.abs((pred_horizon - target_horizon) / (target_horizon + 1e-7))) * 100
|
||||
rmse = np.sqrt(np.mean(np.square(pred_horizon - target_horizon)))
|
||||
|
||||
# Calcola il coefficiente di correlazione
|
||||
corr = np.corrcoef(pred_horizon.flatten(), target_horizon.flatten())[0, 1]
|
||||
|
||||
# Salva le metriche
|
||||
accuracy_metrics[f'horizon_{horizon}'] = {
|
||||
'mae': mae,
|
||||
'mape': mape,
|
||||
'rmse': rmse,
|
||||
'correlation': corr
|
||||
}
|
||||
|
||||
print(f"\nMetriche per orizzonte {horizon}:")
|
||||
print(f"MAE: {mae:.4f}")
|
||||
print(f"MAPE: {mape:.2f}%")
|
||||
print(f"RMSE: {rmse:.4f}")
|
||||
print(f"Correlazione: {corr:.4f}")
|
||||
|
||||
return accuracy_metrics
|
||||
|
||||
|
||||
def compute_confidence_intervals(
|
||||
predictions: np.ndarray,
|
||||
alpha: float = 0.05,
|
||||
n_bootstrap: int = 1000
|
||||
) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
|
||||
"""
|
||||
Calcola intervalli di confidenza usando bootstrap.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
predictions : np.ndarray
|
||||
Predizioni del modello
|
||||
alpha : float
|
||||
Livello di significatività
|
||||
n_bootstrap : int
|
||||
Numero di campioni bootstrap
|
||||
|
||||
Returns
|
||||
-------
|
||||
tuple
|
||||
(lower_bound, upper_bound, mean_predictions)
|
||||
"""
|
||||
n_samples, n_targets = predictions.shape
|
||||
bootstrap_means = np.zeros((n_bootstrap, n_targets))
|
||||
|
||||
# Bootstrap sampling
|
||||
for i in range(n_bootstrap):
|
||||
indices = np.random.randint(0, n_samples, size=n_samples)
|
||||
bootstrap_sample = predictions[indices]
|
||||
bootstrap_means[i] = np.mean(bootstrap_sample, axis=0)
|
||||
|
||||
# Calcola intervalli di confidenza
|
||||
lower_percentile = alpha / 2 * 100
|
||||
upper_percentile = (1 - alpha / 2) * 100
|
||||
|
||||
lower_bound = np.percentile(bootstrap_means, lower_percentile, axis=0)
|
||||
upper_bound = np.percentile(bootstrap_means, upper_percentile, axis=0)
|
||||
mean_predictions = np.mean(predictions, axis=0)
|
||||
|
||||
# Calcola intervalli usando t-distribution
|
||||
std_error = np.std(bootstrap_means, axis=0)
|
||||
t_value = stats.t.ppf(1 - alpha / 2, df=n_samples - 1)
|
||||
margin_error = t_value * std_error
|
||||
|
||||
print("\nIntervalli di Confidenza:")
|
||||
for i in range(n_targets):
|
||||
print(f"\nTarget {i + 1}:")
|
||||
print(f"Media: {mean_predictions[i]:.4f}")
|
||||
print(f"Intervallo: [{lower_bound[i]:.4f}, {upper_bound[i]:.4f}]")
|
||||
print(f"Margine di errore: ±{margin_error[i]:.4f}")
|
||||
|
||||
return lower_bound, upper_bound, mean_predictions
|
||||
Reference in new issue
Block a user