forked from Eshe/competition-vd
128 lines
5.6 KiB
Python
Executable File
128 lines
5.6 KiB
Python
Executable File
import pandas as pd
|
|
import numpy as np
|
|
from sklearn.model_selection import train_test_split
|
|
from typing import Dict, Any
|
|
|
|
from data_wizardry.data_ingestion import DataIngester
|
|
from data_wizardry.feature_craft import FeatureCrafter
|
|
from model_zoo.deep_learner import DeepLearner
|
|
from model_zoo.ensemble_factory import EnsembleFactory
|
|
from model_zoo.experimental_net import ExperimentalNet
|
|
from scoring_magic.cvss_engine import CVSSEngine
|
|
from scoring_magic.risk_oracle import RiskOracle
|
|
from experiment_lab.model_arena import ModelArena
|
|
from experiment_lab.hyper_tuner import HyperTuner
|
|
from insight_generator.plot_master import PlotMaster
|
|
from insight_generator.report_builder import ReportBuilder
|
|
from utils.config_loader import ConfigLoader
|
|
from utils.performance_tracker import PerformanceTracker
|
|
|
|
class CVSSEvaluator:
|
|
def __init__(self, config_path: str = 'config.yaml'):
|
|
self.config = ConfigLoader(config_path)
|
|
self.performance_tracker = PerformanceTracker()
|
|
self.plot_master = PlotMaster()
|
|
self.report_builder = ReportBuilder()
|
|
self.cvss_engine = CVSSEngine()
|
|
self.risk_oracle = RiskOracle()
|
|
|
|
@performance_tracker.track_execution_time
|
|
def load_and_preprocess_data(self) -> Dict[str, pd.DataFrame]:
|
|
data_ingester = DataIngester(self.config.get('data_paths'))
|
|
data = data_ingester.process()
|
|
|
|
feature_crafter = FeatureCrafter()
|
|
processed_data = {}
|
|
for split, df in data.items():
|
|
X = feature_crafter.fit_transform(df) if split == 'train' else feature_crafter.transform(df)
|
|
y = df['baseScore']
|
|
processed_data[split] = {'X': X, 'y': y}
|
|
|
|
return processed_data
|
|
|
|
@performance_tracker.track_execution_time
|
|
def train_models(self, data: Dict[str, Dict[str, pd.DataFrame]]) -> Dict[str, Any]:
|
|
X_train, y_train = data['train']['X'], data['train']['y']
|
|
X_val, y_val = data['val']['X'], data['val']['y']
|
|
|
|
model_arena = ModelArena(X_train, y_train, X_val, y_val)
|
|
model_arena.train_and_evaluate()
|
|
|
|
best_model_name = model_arena.get_best_model()
|
|
best_model = model_arena.models[best_model_name]
|
|
|
|
return {
|
|
'best_model': best_model,
|
|
'best_model_name': best_model_name,
|
|
'all_results': model_arena.get_results_dataframe()
|
|
}
|
|
|
|
@performance_tracker.track_execution_time
|
|
def hyperparameter_tuning(self, data: Dict[str, Dict[str, pd.DataFrame]]) -> Dict[str, Any]:
|
|
X_train, y_train = data['train']['X'], data['train']['y']
|
|
hyper_tuner = HyperTuner(X_train, y_train)
|
|
|
|
rf_results = hyper_tuner.tune_random_forest()
|
|
gb_results = hyper_tuner.tune_gradient_boosting()
|
|
exp_results = hyper_tuner.tune_experimental_net()
|
|
|
|
return {
|
|
'random_forest': rf_results,
|
|
'gradient_boosting': gb_results,
|
|
'experimental_net': exp_results
|
|
}
|
|
|
|
@performance_tracker.track_execution_time
|
|
def evaluate_model(self, model: Any, data: Dict[str, Dict[str, pd.DataFrame]]) -> Dict[str, float]:
|
|
X_test, y_test = data['test']['X'], data['test']['y']
|
|
y_pred = model.predict(X_test)
|
|
|
|
self.performance_tracker.calculate_metrics(y_test, y_pred)
|
|
return self.performance_tracker.get_metrics()
|
|
|
|
def generate_visualizations(self, data: Dict[str, Dict[str, pd.DataFrame]], model: Any, results: pd.DataFrame):
|
|
X_test, y_test = data['test']['X'], data['test']['y']
|
|
y_pred = model.predict(X_test)
|
|
|
|
self.plot_master.plot_feature_importance(model.get_feature_importance(X_test))
|
|
self.plot_master.plot_prediction_vs_actual(y_test, y_pred)
|
|
self.plot_master.plot_error_distribution(y_test, y_pred)
|
|
self.plot_master.plot_model_comparison(results)
|
|
|
|
def generate_report(self, model: Any, data: Dict[str, Dict[str, pd.DataFrame]], results: pd.DataFrame, tuning_results: Dict[str, Any]):
|
|
self.report_builder.add_model_performance(results.to_dict())
|
|
self.report_builder.add_feature_importance(model.get_feature_importance(data['test']['X']))
|
|
self.report_builder.add_prediction_stats(data['test']['y'], model.predict(data['test']['X']))
|
|
|
|
for plot_func in [self.plot_master.plot_feature_importance,
|
|
self.plot_master.plot_prediction_vs_actual,
|
|
self.plot_master.plot_error_distribution,
|
|
self.plot_master.plot_model_comparison]:
|
|
self.report_builder.add_plot(plot_func, *plot_func.__code__.co_varnames[1:])
|
|
|
|
return self.report_builder.generate_html_report()
|
|
|
|
def run(self):
|
|
data = self.load_and_preprocess_data()
|
|
model_results = self.train_models(data)
|
|
best_model = model_results['best_model']
|
|
all_results = model_results['all_results']
|
|
|
|
tuning_results = self.hyperparameter_tuning(data)
|
|
|
|
evaluation_metrics = self.evaluate_model(best_model, data)
|
|
self.generate_visualizations(data, best_model, all_results)
|
|
|
|
report = self.generate_report(best_model, data, all_results, tuning_results)
|
|
|
|
print(f"Best Model: {model_results['best_model_name']}")
|
|
print(f"Evaluation Metrics: {evaluation_metrics}")
|
|
print(f"Performance Report:\n{self.performance_tracker.generate_performance_report()}")
|
|
print(f"Full Report saved as 'cvss_evaluation_report.html'")
|
|
|
|
with open('cvss_evaluation_report.html', 'w') as f:
|
|
f.write(report)
|
|
|
|
if __name__ == "__main__":
|
|
evaluator = CVSSEvaluator()
|
|
evaluator.run() |