Add Finam importer

actions/finam.py

@@ -0,0 +1,68 @@
+import os
+from pathlib import Path
+from typing import Any, Tuple
+
+import pandas as pd
+
+from common import Cmd, Symbol, p, tzinfo
+from fs import Block, Store
+
+
+class Import(Cmd):
+    def run(self, *args: str, **kwargs: Any) -> Tuple[int | None, str | Exception | None]:
+        symbols: dict[str, Symbol] | None = kwargs.get("symbols")
+        if not symbols:
+            return None, None
+        path: str | os.PathLike[str] | None = kwargs.get("path")
+        path = path if path is not None else ""
+        store = Store(path)
+        for arg in args:
+            sym, err = self.__parse_arg(arg)
+            if err is not None:
+                return 1, err
+            if sym.lower() not in symbols:
+                p(f"Skipping {arg}")
+                continue
+            p(f"Processing {arg}... ", end="")
+            err = self.__process_arg(arg, symbols[sym.lower()], store)
+            if isinstance(err, FileNotFoundError):
+                p("file not found")
+                return 1, None
+            if err is not None:
+                p()
+                return 2, err
+            p("done.")
+        return None, None
+
+    @staticmethod
+    def __parse_arg(arg: str) -> Tuple[str, str | Exception | None]:
+        symbol, sep, _ = Path(arg).stem.partition("_")
+        if sep != "_":
+            return "", f"invalid arg: {arg}"
+        return symbol, None
+
+    @staticmethod
+    def __process_arg(arg: str, symbol: Symbol, store: Store) -> str | Exception | None:
+        try:
+            df = pd.read_csv(arg, sep="\t", dtype="str", skiprows=1, header=None, usecols=range(8))
+        except OSError as e:
+            return e
+        if symbol.time is None:
+            df[0] = pd.to_datetime(df[0] + " " + df[1], format="%Y.%m.%d %H:%M:%S", utc=True)
+        else:
+            df[0] = pd.to_datetime(df[0] + " " + df[1], format="%Y.%m.%d %H:%M:%S")
+            df[0] = df[0].dt.tz_localize(tzinfo(symbol.time))
+        df = df.drop(columns=[1, 6])
+        for date, gf in df.groupby(df[0].dt.date):
+            err = store.put(Block(symbol.name, symbol.market, date, gf))
+            if err is not None:
+                return err
+        return None
+
+
+def getcmd(name: str) -> Tuple[Cmd | None, str | None]:
+    match name:
+        case "import":
+            return Import(), None
+        case _:
+            return None, f"command {name} not found in module {__name__}"

actions/high.py

@@ -1,7 +1,6 @@
-import datetime
 import os
 from pathlib import Path
-from typing import Any, Tuple, cast
+from typing import Any, Tuple
 
 import pandas as pd
 
@@ -51,7 +50,7 @@ def __process_arg(arg: str, symbol: Symbol, store: Store) -> str | Exception | N
             df[0] = pd.to_datetime(df[0], format="%Y%m%d %H:%M")
             df[0] = df[0].dt.tz_localize(tzinfo(symbol.time))
         for date, gf in df.groupby(df[0].dt.date):
-            err = store.put(Block(symbol.name, symbol.market, cast(datetime.date, date), gf))
+            err = store.put(Block(symbol.name, symbol.market, date, gf))
             if err is not None:
                 return err
         return None

actions/kraken.py

@@ -1,7 +1,6 @@
-import datetime
 import os
 import zipfile
-from typing import IO, Any, Tuple, cast
+from typing import IO, Any, Tuple
 
 import pandas as pd
 
@@ -56,7 +55,7 @@ def __process_file(file: IO[bytes], symbol: Symbol, store: Store) -> str | Excep
         if symbol.time is not None:
             df[0] = df[0].dt.tz_convert(tzinfo(symbol.time))
         for date, gf in df.groupby(df[0].dt.date):
-            err = store.put(Block(symbol.name, symbol.market, cast(datetime.date, date), gf))
+            err = store.put(Block(symbol.name, symbol.market, date, gf))
             if err is not None:
                 return err
         return None

conf.py

@@ -102,6 +102,8 @@ def _walk_symbols(
             else:
                 symbol.start = symbol.start.replace(tzinfo=datetime.timezone.utc)
         if name is not None:
+            if type(name) != str:
+                raise TypeError
             yield replace(symbol, name=name)
         elif symbols is not None:
             if not isinstance(symbols, Iterable):
@@ -132,6 +134,8 @@ def _walk_actions(
                 case _:
                     raise Error(f"unexpected key: {k}")
         if name is not None:
+            if type(name) != str:
+                raise TypeError
             yield replace(action, name=name)
         elif actions is not None:
             if not isinstance(actions, Iterable):