✅ Add kpss and ocsb tests

Author: LuizFCDuarte

test/datasets/kpss_results_datasets.json

@@ -0,0 +1,46 @@
+{
+    "GDPC1.csv": {
+        "test_stat": 2.90119724088986,
+        "p_value": 0.01,
+        "lags_used": 11,
+        "critical_values": {
+            "10%": 0.347,
+            "5%": 0.463,
+            "2.5%": 0.574,
+            "1%": 0.739
+        }
+    },
+    "NROU.csv": {
+        "test_stat": 2.0958404238268487,
+        "p_value": 0.01,
+        "lags_used": 11,
+        "critical_values": {
+            "10%": 0.347,
+            "5%": 0.463,
+            "2.5%": 0.574,
+            "1%": 0.739
+        }
+    },
+    "airpassengers.csv": {
+        "test_stat": 2.0131256386303322,
+        "p_value": 0.01,
+        "lags_used": 9,
+        "critical_values": {
+            "10%": 0.347,
+            "5%": 0.463,
+            "2.5%": 0.574,
+            "1%": 0.739
+        }
+    },
+    "log_airpassengers.csv": {
+        "test_stat": 2.1181889701236964,
+        "p_value": 0.01,
+        "lags_used": 9,
+        "critical_values": {
+            "10%": 0.347,
+            "5%": 0.463,
+            "2.5%": 0.574,
+            "1%": 0.739
+        }
+    }
+}

test/datasets/kpss_script.py

@@ -1,4 +1,4 @@
-import json
+import json, os
 import numpy as np
 import pandas as pd
 from statsmodels.tsa.stattools import kpss
@@ -28,18 +28,61 @@ def save_json(data, filename):
     with open(filename, 'w') as f:
         json.dump(data, f, indent=4)
 
+def read_datasets_and_apply_kpss_test():
+    # Directory containing the datasets
+    DATASETS_DIR = os.path.dirname(os.path.abspath(__file__))
+    DATASETS_DIR = os.path.abspath(os.path.join(DATASETS_DIR, '../../datasets'))
+
+    files = [
+        ('GDPC1.csv', ';', 'GDPC1'),
+        ('NROU.csv', ';', 'NROU'),
+        ('airpassengers.csv', ',', 'value'),
+        ('log_airpassengers.csv', ',', 'value'),
+    ]
+
+    results = {}
+
+    for fname, delim, value_col in files:
+        if "log" in fname:
+            use_log = True
+            fname = fname.split("_")[1]
+        else:
+            use_log = False
+        path = os.path.join(DATASETS_DIR, fname)
+        df = pd.read_csv(path, delimiter=delim)
+        # Handle possible quoted headers
+        if value_col not in df.columns:
+            # Try stripping quotes
+            df.columns = [c.strip('"') for c in df.columns]
+        series = df[value_col].astype(float)
+        if use_log:
+            series = np.log(series)
+        result = kpss(series)
+        if use_log:
+            fname = "log_" + fname
+        results[fname] = {
+            'test_stat': result[0],
+            'p_value': result[1],
+            'lags_used': result[2],
+            'critical_values': result[3]
+        }
+
+    # Save the results as json
+    save_json(results, "kpss_results_datasets.json")
+
 if __name__ == "__main__":
-series_data = {}
-kpss_results_data = {}
-for i in range(1, 11):
-    if i % 2 == 0:
-        series = generate_nonstationary_series(trend=True, seasonality=(i % 4 == 0))
-        label = f"nonstationary_series_{i}"
-    else:
-        series = generate_stationary_series()
-        label = f"stationary_series_{i}"
-    series_data[label] = series
-    kpss_results_data[label] = perform_kpss_test(series)
-save_json(series_data, "time_series.json")
-save_json(kpss_results_data, "kpss_results.json")
+    # series_data = {}
+    # kpss_results_data = {}
+    # for i in range(1, 11):
+    #     if i % 2 == 0:
+    #         series = generate_nonstationary_series(trend=True, seasonality=(i % 4 == 0))
+    #         label = f"nonstationary_series_{i}"
+    #     else:
+    #         series = generate_stationary_series()
+    #         label = f"stationary_series_{i}"
+    # series_data[label] = series
+    # kpss_results_data[label] = perform_kpss_test(series)
+    # save_json(series_data, "time_series.json")
+    # save_json(kpss_results_data, "kpss_results.json")
+    read_datasets_and_apply_kpss_test()
 print("Files saved: time_series.json, kpss_results.json")

test/models/sarima_predict.jl

@@ -155,7 +155,7 @@ end
             exog = exogSeries,
             seasonality = 12,
             objectiveFunction = "lasso",
-            seasonalIntegrationTest = "ch"
+            seasonalIntegrationTest = "ocsb"
         )
         forecastExog = Sarimax.predict!(modelExog; stepsAhead = length(testingSet))
         mapeExog = MAPE(testingSet, forecastExog)

test/test_statistical_tests.jl

@@ -66,6 +66,30 @@ using JSON
         @test_throws ArgumentError Sarimax.kpss_test(data, regression=:invalid)
     end
 
+    @testset "Test in airpassengers" begin
+        kpss_datasets = JSON.parsefile(joinpath(@__DIR__, "datasets", "kpss_results_datasets.json"))
+        airpassengers = loadDataset(AIR_PASSENGERS)
+        kpss_result = Sarimax.kpss_test(values(airpassengers);regression=:c)
+        @test isapprox(kpss_result["test_statistic"], kpss_datasets["airpassengers.csv"]["test_stat"], atol=5e-3)
+        @test kpss_result["p_value"] == kpss_datasets["airpassengers.csv"]["p_value"]
+
+        airpassengersLog = log.(values(airpassengers))
+        kpss_result = Sarimax.kpss_test(airpassengersLog;regression=:c)
+        @test isapprox(kpss_result["test_statistic"], kpss_datasets["log_airpassengers.csv"]["test_stat"], atol=5e-3)
+        @test kpss_result["p_value"] == kpss_datasets["log_airpassengers.csv"]["p_value"]
+
+        gdpc1 = loadDataset(GDPC1)
+        kpss_result = Sarimax.kpss_test(values(gdpc1);regression=:c)
+        @test isapprox(kpss_result["test_statistic"], kpss_datasets["GDPC1.csv"]["test_stat"], atol=5e-3)
+        @test kpss_result["p_value"] == kpss_datasets["GDPC1.csv"]["p_value"]
+
+        nrou = loadDataset(NROU)
+        kpss_result = Sarimax.kpss_test(values(nrou);regression=:c)
+        @test isapprox(kpss_result["test_statistic"], kpss_datasets["NROU.csv"]["test_stat"], atol=5e-3)
+        @test kpss_result["p_value"] == kpss_datasets["NROU.csv"]["p_value"]
+    end
+
+
     @testset "Lag Selection" begin
         Random.seed!(123)
         data = randn(100)