[quantile regression project] add exp params. refactor annual maxima simulations

extreme_fit/distribution/exp_params.py

+from extreme_fit.distribution.abstract_params import AbstractParams
+
+
+class ExpParams(object):
+    PARAM_NAMES = [AbstractParams.SCALE]
+
+    def __init__(self, scale: float):
+        self.scale = scale
+

extreme_fit/model/margin_model/linear_margin_model/linear_margin_model.py

@@ -9,9 +9,9 @@ class LinearMarginModel(ParametricMarginModel):
     @classmethod
     def from_coef_list(cls, coordinates, gev_param_name_to_coef_list, **kwargs):
         params = {}
-        for gev_param_name in GevParams.PARAM_NAMES:
-            for idx, coef in enumerate(gev_param_name_to_coef_list[gev_param_name], -1):
-                params[(gev_param_name, idx)] = coef
+        for param_name, coef_list in gev_param_name_to_coef_list.items():
+            for idx, coef in enumerate(coef_list, -1):
+                params[(param_name, idx)] = coef
         return cls(coordinates, params_sample=params, params_start_fit=params, **kwargs)
 
     def load_margin_functions(self, gev_param_name_to_dims=None):

projects/quantile_regression_vs_evt/annual_maxima_simulation/abstract_annual_maxima_simulation.py

+from collections import OrderedDict
+from typing import List, Dict
+
+import numpy as np
+from cached_property import cached_property
+
+from extreme_fit.estimator.quantile_estimator.abstract_quantile_estimator import AbstractQuantileEstimator
+from extreme_fit.model.margin_model.abstract_margin_model import AbstractMarginModel
+from projects.quantile_regression_vs_evt.AbstractSimulation import AbstractSimulation
+from spatio_temporal_dataset.spatio_temporal_observations.abstract_spatio_temporal_observations import \
+    AbstractSpatioTemporalObservations
+
+
+class AnnualMaximaSimulation(AbstractSimulation):
+
+    @property
+    def observations_class(self):
+        raise NotImplementedError
+
+    @cached_property
+    def time_series_lengths_to_margin_model(self) -> Dict[int, AbstractMarginModel]:
+        d = OrderedDict()
+        for length in self.time_series_lengths:
+            coordinates = self.time_series_length_to_coordinates[length]
+            d[length] = self.create_model(coordinates)
+        return d
+
+    def create_model(self, coordinates):
+        raise NotImplementedError
+
+    def generate_all_observation(self, nb_time_series, length) -> List[AbstractSpatioTemporalObservations]:
+        coordinates = self.time_series_length_to_coordinates[length]
+        margin_model = self.time_series_lengths_to_margin_model[length]
+        return [self.observations_class.from_sampling(nb_obs=1, coordinates=coordinates, margin_model=margin_model)
+                for _ in range(nb_time_series)]
+
+    def compute_errors(self, length: int, estimators: List[AbstractQuantileEstimator]):
+        coordinates = self.time_series_length_to_coordinates[length]
+        last_coordinate = coordinates.coordinates_values()[-1]
+        # Compute true value
+        margin_model = self.time_series_lengths_to_margin_model[length]
+        true_quantile = margin_model.margin_function_sample.get_gev_params(last_coordinate).quantile(self.quantile)
+        # Compute estimated values
+        estimated_quantiles = [estimator.function_from_fit.get_quantile(last_coordinate) for estimator in estimators]
+        return 100 * np.abs(np.array(estimated_quantiles) - true_quantile) / true_quantile

projects/quantile_regression_vs_evt/annual_maxima_simulation/daily_exp_simulation.py

+from extreme_fit.distribution.gev.gev_params import GevParams
+from extreme_fit.model.margin_model.linear_margin_model.abstract_temporal_linear_margin_model import \
+    TemporalMarginFitMethod
+from extreme_fit.model.margin_model.linear_margin_model.temporal_linear_margin_models import StationaryTemporalModel
+from projects.quantile_regression_vs_evt.annual_maxima_simulation.abstract_annual_maxima_simulation import \
+    AnnualMaximaSimulation
+from spatio_temporal_dataset.spatio_temporal_observations.annual_maxima_observations import DailyExpAnnualMaxima
+
+
+class DailyExpSimulation(AnnualMaximaSimulation):
+
+    @property
+    def observations_class(self):
+        return DailyExpAnnualMaxima
+
+
+class StationaryExpSimulation(DailyExpSimulation):
+
+    def create_model(self, coordinates):
+        gev_param_name_to_coef_list = {
+            GevParams.SCALE: [1],
+        }
+        return StationaryTemporalModel.from_coef_list(coordinates, gev_param_name_to_coef_list,
+                                                      fit_method=TemporalMarginFitMethod.extremes_fevd_mle)
+
+

projects/quantile_regression_vs_evt/GevSimulation.py → projects/quantile_regression_vs_evt/annual_maxima_simulation/gev_simulation.py

@@ -12,40 +12,20 @@ from extreme_fit.model.margin_model.linear_margin_model.abstract_temporal_linear
 from extreme_fit.model.margin_model.linear_margin_model.temporal_linear_margin_models import StationaryTemporalModel, \
     NonStationaryLocationTemporalModel
 from projects.quantile_regression_vs_evt.AbstractSimulation import AbstractSimulation
+from projects.quantile_regression_vs_evt.annual_maxima_simulation.abstract_annual_maxima_simulation import \
+    AnnualMaximaSimulation
 from spatio_temporal_dataset.spatio_temporal_observations.abstract_spatio_temporal_observations import \
     AbstractSpatioTemporalObservations
 from spatio_temporal_dataset.spatio_temporal_observations.annual_maxima_observations import MarginAnnualMaxima
 
 
-class GevSimulation(AbstractSimulation):
 
-    @cached_property
-    def time_series_lengths_to_margin_model(self) -> Dict[int, AbstractMarginModel]:
-        d = OrderedDict()
-        for length in self.time_series_lengths:
-            coordinates = self.time_series_length_to_coordinates[length]
-            d[length] = self.create_model(coordinates)
-        return d
 
-    def create_model(self, coordinates):
-        raise NotImplementedError
-
-    def generate_all_observation(self, nb_time_series, length) -> List[AbstractSpatioTemporalObservations]:
-        coordinates = self.time_series_length_to_coordinates[length]
-        margin_model = self.time_series_lengths_to_margin_model[length]
-        return [MarginAnnualMaxima.from_sampling(nb_obs=1, coordinates=coordinates, margin_model=margin_model)
-                for _ in range(nb_time_series)]
-
-    def compute_errors(self, length: int, estimators: List[AbstractQuantileEstimator]):
-        coordinates = self.time_series_length_to_coordinates[length]
-        last_coordinate = coordinates.coordinates_values()[-1]
-        # Compute true value
-        margin_model = self.time_series_lengths_to_margin_model[length]
-        true_quantile = margin_model.margin_function_sample.get_gev_params(last_coordinate).quantile(self.quantile)
-        # Compute estimated values
-        estimated_quantiles = [estimator.function_from_fit.get_quantile(last_coordinate) for estimator in estimators]
-        return 100 * np.abs(np.array(estimated_quantiles) - true_quantile) / true_quantile
+class GevSimulation(AnnualMaximaSimulation):
 
+    @property
+    def observations_class(self):
+        return MarginAnnualMaxima
 
 class StationarySimulation(GevSimulation):
 

projects/quantile_regression_vs_evt/main_quantile_regression.py

-from extreme_fit.model.margin_model.linear_margin_model.temporal_linear_margin_models import StationaryTemporalModel, \
+from extreme_fit.model.margin_model.linear_margin_model.temporal_linear_margin_models import \
     NonStationaryLocationTemporalModel, NonStationaryLocationGumbelModel
-from extreme_fit.model.quantile_model.quantile_regression_model import ConstantQuantileRegressionModel, \
-    TemporalCoordinatesQuantileRegressionModel
-from projects.quantile_regression_vs_evt.GevSimulation import StationarySimulation, \
+from extreme_fit.model.quantile_model.quantile_regression_model import TemporalCoordinatesQuantileRegressionModel
+from projects.quantile_regression_vs_evt.annual_maxima_simulation.gev_simulation import \
     NonStationaryLocationGumbelSimulation, NonStationaryLocationGevSimulation
 from spatio_temporal_dataset.coordinates.transformed_coordinates.transformation.abstract_transformation import \
     CenteredScaledNormalization, IdentityTransformation
-from spatio_temporal_dataset.coordinates.transformed_coordinates.transformation.uniform_normalization import \
-    BetweenZeroAndOneNormalization
 
 nb_time_series = 10
 quantile = 0.98

spatio_temporal_dataset/spatio_temporal_observations/annual_maxima_observations.py

@@ -30,6 +30,22 @@ class MarginAnnualMaxima(AnnualMaxima):
         return cls(df_maxima_gev=df_maxima_gev)
 
 
+class DailyExpAnnualMaxima(AnnualMaxima):
+
+    @classmethod
+    def from_sampling(cls, nb_obs: int, coordinates: AbstractCoordinates,
+                      margin_model: AbstractMarginModel):
+        pass
+
+
+class DailyExp(AbstractSpatioTemporalObservations):
+
+    @classmethod
+    def from_sampling(cls, nb_obs: int, coordinates: AbstractCoordinates,
+                      margin_model: AbstractMarginModel):
+        pass
+
+
 class MaxStableAnnualMaxima(AnnualMaxima):
 
     @classmethod

test/test_projects/test_quantile_regression/test_gev_simulations.py → test/test_projects/test_quantile_regression/test_annual_maxima_simulations.py

@@ -4,7 +4,8 @@ from extreme_fit.model.margin_model.linear_margin_model.temporal_linear_margin_m
     NonStationaryLocationTemporalModel
 from extreme_fit.model.quantile_model.quantile_regression_model import ConstantQuantileRegressionModel, \
     TemporalCoordinatesQuantileRegressionModel
-from projects.quantile_regression_vs_evt.GevSimulation import GevSimulation, StationarySimulation, \
+from projects.quantile_regression_vs_evt.annual_maxima_simulation.daily_exp_simulation import StationaryExpSimulation
+from projects.quantile_regression_vs_evt.annual_maxima_simulation.gev_simulation import StationarySimulation, \
     NonStationaryLocationGumbelSimulation
 
 
@@ -19,9 +20,18 @@ class TestGevSimulations(unittest.TestCase):
     def test_non_stationary_run(self):
         simulation = NonStationaryLocationGumbelSimulation(nb_time_series=1, quantile=0.5, time_series_lengths=[50, 60],
                                                            model_classes=[NonStationaryLocationTemporalModel,
-                                                                    TemporalCoordinatesQuantileRegressionModel])
+                                                                          TemporalCoordinatesQuantileRegressionModel])
         simulation.plot_error_for_last_year_quantile(self.DISPLAY)
 
 
+# class TestExpSimulations(unittest.TestCase):
+#     DISPLAY = True
+#
+#     def test_stationary_run(self):
+#         simulation = StationaryExpSimulation(nb_time_series=1, quantile=0.5, time_series_lengths=[50, 60],
+#                                              model_classes=[StationaryTemporalModel, ConstantQuantileRegressionModel])
+#         simulation.plot_error_for_last_year_quantile(self.DISPLAY)
+
+
 if __name__ == '__main__':
     unittest.main()