added downloading of sunspots, started work on downloading of geomagnetic indexes and CBOE data

Author: wegar-2

moddata/_utils.py

@@ -13,7 +13,9 @@
 Dataset: TypeAlias = Literal[
     "bankchurn",
     "btc",
-    "pl_banking_stocks"
+    "pl_banking_stocks",
+    "sunspots",
+    "geomagnetic_indexes"
 ]
 
 
@@ -65,15 +67,21 @@ def _load_pl_banking_stocks() -> pd.DataFrame:
     ))
 
 
+def _load_sunspots() -> pd.DataFrame:
+    return pd.read_parquet(str(
+        resources.files('moddata.data').joinpath('sunspots.parquet')
+    ))
+
+
 def load_data(dataset: Dataset) -> pd.DataFrame | None:
     if dataset == "bankchurn":
         return _load_bankchurn()
     if dataset == "btc":
         return _load_btc()
     if dataset == "pl_banking_stocks":
         return _load_pl_banking_stocks()
+    if dataset == "sunspots":
+        raise _load_sunspots()
+    if dataset == "geomagnetic_indexes":
+        raise Exception()
     raise ValueError(f"Encountered invalid dataset name: {dataset}")
-
-
-if __name__ == "__main__":
-    _load_btc()

moddata/data/sunspots.parquet

@@ -0,0 +1,7 @@
+import pandas as pd
+
+
+class DownloadCboeDataExtractor:
+
+    def extract(self):
+        pass

moddata/extractor/download_cboe_data_extractor.py

@@ -0,0 +1,15 @@
+from typing import Final
+
+import pandas as pd
+
+
+class DownloadGeomagneticIndexExtractor:
+
+    _DATA_URL: Final[str] = "https://kp.gfz.de/app/files/Kp_ap_Ap_SN_F107_since_1932.txt"
+
+    def extract(self) -> pd.DataFrame:
+        pass
+
+
+if __name__ == '__main__':
+    DownloadGeomagneticIndexExtractor()

moddata/extractor/download_geomagnetic_index_extractor.py

@@ -0,0 +1,48 @@
+import logging
+from typing import Final
+
+import numpy as np
+import pandas as pd
+
+logger = logging.getLogger(__name__)
+
+
+class DownloadSunspotsExtractor:
+
+    _DAILY_TOTAL_SUNSPOT_NUMBER_URL: Final[str] = "https://www.sidc.be/SILSO/INFO/sndtotcsv.php"
+
+    def extract(self) -> pd.DataFrame:
+        data = pd.read_csv(
+            self._DAILY_TOTAL_SUNSPOT_NUMBER_URL,
+            delimiter=";",
+            decimal=".",
+            names=[
+                "year", "month", "day", "yearfrac_date",
+                "daily_sunspots_number",
+                "daily_std_across_stations",
+                "obs_num", "is_definitive"
+            ],
+            na_values=-1
+        )
+        data = data[["year", "month", "day", "daily_sunspots_number"]]
+        data["day"] = (
+                data["year"].astype(str) + "-" +
+                data["month"].apply(lambda x: f"{x:02}") + "-" +
+                data["day"].apply( lambda x: f"{x:02}")
+        )
+        data = data[["day", "daily_sunspots_number"]]
+        data["daily_sunspots_number"] = np.where(
+            data["daily_sunspots_number"] == -1,
+            np.nan,
+            data["daily_sunspots_number"]
+        )
+        return data
+
+
+if __name__ == "__main__":
+    from pathlib import Path
+    data = DownloadSunspotsExtractor().extract()
+    data.to_parquet(
+        Path(__file__).parent.parent / "data" / "sunspots.parquet"
+    )
+    print("halt")