Add fred data source (#11)

Author: lsbardel

quantflow/cli/app.py

@@ -1,31 +1,40 @@
 import os
 from dataclasses import dataclass, field
 from typing import Any
-
+from functools import partial
 import click
 from prompt_toolkit import PromptSession
 from prompt_toolkit.history import FileHistory
 from rich.console import Console
 from rich.text import Text
+from .commands import fred, stocks, vault
+from quantflow.data.vault import Vault
+from quantflow.data.fmp import FMP
+from quantflow.data.fred import Fred
 
-
-from . import settings, commands
+from . import settings
 
 
 @click.group()
 def qf() -> None:
     pass
 
 
-qf.add_command(commands.exit)
-qf.add_command(commands.profile)
-qf.add_command(commands.search)
-qf.add_command(commands.chart)
+@qf.command()
+def exit() -> None:
+    """Exit the program"""
+    raise click.Abort()
+
+
+qf.add_command(vault.vault)
+qf.add_command(stocks.stocks)
+qf.add_command(fred.fred)
 
 
 @dataclass
 class QfApp:
     console: Console = field(default_factory=Console)
+    vault: Vault = field(default_factory=partial(Vault, settings.VAULT_FILE_PATH))
 
     def __call__(self) -> None:
         os.makedirs(settings.SETTINGS_DIRECTORY, exist_ok=True)
@@ -70,3 +79,15 @@ def handle_command(self, text: str) -> None:
             self.error(e)
         except click.exceptions.UsageError as e:
             self.error(e)
+
+    def fmp(self) -> FMP:
+        if key := self.vault.get("fmp"):
+            return FMP(key=key)
+        else:
+            raise click.UsageError("No FMP API key found")
+
+    def fred(self) -> Fred:
+        if key := self.vault.get("fred"):
+            return Fred(key=key)
+        else:
+            raise click.UsageError("No FRED API key found")

quantflow/cli/commands/__init__.py

@@ -0,0 +1,68 @@
+from __future__ import annotations
+
+import click
+import asyncio
+import pandas as pd
+from typing import TYPE_CHECKING
+from fluid.utils.data import compact_dict
+from fluid.utils.http_client import HttpResponseError
+from ccy.cli.console import df_to_rich
+from quantflow.data.fmp import FMP
+from .stocks import from_context
+
+
+FREQUENCIES = tuple(FMP().historical_frequencies())
+
+if TYPE_CHECKING:
+    from quantflow.cli.app import QfApp
+
+
+@click.group(invoke_without_command=True)
+@click.pass_context
+def fred(ctx: click.Context) -> None:
+    """Federal Reserve of St. Louis data"""
+    if ctx.invoked_subcommand is None:
+        app = from_context(ctx)
+        app.print("Welcome to FRED data commands!")
+        app.print(ctx.get_help())
+
+
+@fred.command()
+@click.pass_context
+@click.argument("category-id", required=False)
+def subcategories(ctx: click.Context, category_id: str | None = None) -> None:
+    """List subcategories for a Fred category"""
+    app = from_context(ctx)
+    try:
+        data = asyncio.run(get_subcategories(app, category_id))
+    except HttpResponseError as e:
+        app.error(e)
+    else:
+        df = pd.DataFrame(data["categories"], columns=["id", "name"])
+        app.print(df_to_rich(df))
+
+
+@fred.command()
+@click.pass_context
+@click.argument("category-id")
+def series(ctx: click.Context, category_id: str) -> None:
+    """List series for a Fred category"""
+    app = from_context(ctx)
+    try:
+        data = asyncio.run(get_series(app, category_id))
+    except HttpResponseError as e:
+        app.error(e)
+    else:
+        app.print(data)
+        # df = pd.DataFrame(data["categories"], columns=["id", "name"])
+        # app.print(df_to_rich(df))
+
+
+async def get_subcategories(app: QfApp, category_id: str | None) -> dict:
+    async with app.fred() as cli:
+        return await cli.subcategories(params=compact_dict(category_id=category_id))
+
+
+async def get_series(app: QfApp, category_id: str) -> dict:
+    async with app.fred() as cli:
+        return await cli.series(params=compact_dict(category_id=category_id))

quantflow/cli/commands/fred.py

@@ -18,19 +18,23 @@ def from_context(ctx: click.Context) -> QfApp:
     return ctx.obj  # type: ignore
 
 
-@click.command()
-def exit() -> None:
-    """Exit the program"""
-    raise click.Abort()
+@click.group(invoke_without_command=True)
+@click.pass_context
+def stocks(ctx: click.Context) -> None:
+    """Stocks commands"""
+    if ctx.invoked_subcommand is None:
+        app = from_context(ctx)
+        app.print("Welcome to the stocks commands!")
+        app.print(ctx.get_help())
 
 
-@click.command()
+@stocks.command()
 @click.argument("symbol")
 @click.pass_context
 def profile(ctx: click.Context, symbol: str) -> None:
     """Company profile"""
     app = from_context(ctx)
-    data = asyncio.run(get_profile(symbol))
+    data = asyncio.run(get_profile(app, symbol))
     if not data:
         raise click.UsageError(f"Company {symbol} not found - try searching")
     else:
@@ -40,18 +44,18 @@ def profile(ctx: click.Context, symbol: str) -> None:
         app.print(df_to_rich(df))
 
 
-@click.command()
+@stocks.command()
 @click.argument("text")
 @click.pass_context
 def search(ctx: click.Context, text: str) -> None:
     """Search companies"""
     app = from_context(ctx)
-    data = asyncio.run(search_company(text))
+    data = asyncio.run(search_company(app, text))
     df = pd.DataFrame(data, columns=["symbol", "name", "currency", "stockExchange"])
     app.print(df_to_rich(df))
 
 
-@click.command()
+@stocks.command()
 @click.argument("symbol")
 @click.option(
     "-h",
@@ -76,9 +80,13 @@ def search(ctx: click.Context, text: str) -> None:
     default="",
     help="Frequency of data - if not provided it is daily",
 )
-def chart(symbol: str, height: int, length: int, frequency: str) -> None:
+@click.pass_context
+def chart(
+    ctx: click.Context, symbol: str, height: int, length: int, frequency: str
+) -> None:
     """Symbol chart"""
-    df = asyncio.run(get_prices(symbol, frequency))
+    app = from_context(ctx)
+    df = asyncio.run(get_prices(app, symbol, frequency))
     if df.empty:
         raise click.UsageError(
             f"No data for {symbol} - are you sure the symbol exists?"
@@ -87,16 +95,16 @@ def chart(symbol: str, height: int, length: int, frequency: str) -> None:
     print(plot(data, {"height": height}))
 
 
-async def get_prices(symbol: str, frequency: str) -> pd.DataFrame:
-    async with FMP() as cli:
+async def get_prices(app: QfApp, symbol: str, frequency: str) -> pd.DataFrame:
+    async with app.fmp() as cli:
         return await cli.prices(symbol, frequency)
 
 
-async def get_profile(symbol: str) -> list[dict]:
-    async with FMP() as cli:
+async def get_profile(app: QfApp, symbol: str) -> list[dict]:
+    async with app.fmp() as cli:
         return await cli.profile(symbol)
 
 
-async def search_company(text: str) -> list[dict]:
-    async with FMP() as cli:
+async def search_company(app: QfApp, text: str) -> list[dict]:
+    async with app.fmp() as cli:
         return await cli.search(text)

quantflow/cli/commands.py quantflow/cli/commands/stocks.pyquantflow/cli/commands.py renamed to quantflow/cli/commands/stocks.py

@@ -0,0 +1,54 @@
+import click
+from .stocks import from_context
+
+
+API_KEYS = ("fmp", "fred")
+
+
+@click.group()
+def vault() -> None:
+    """Manage vault secrets"""
+    pass
+
+
+@vault.command()
+@click.argument("key", type=click.Choice(API_KEYS))
+@click.argument("value")
+@click.pass_context
+def add(ctx: click.Context, key: str, value: str) -> None:
+    """Add an API key to the vault"""
+    vault = from_context(ctx).vault
+    vault.add(key, value)
+
+
+@vault.command()
+@click.argument("key")
+@click.pass_context
+def delete(ctx: click.Context, key: str) -> None:
+    """Delete an API key from the vault"""
+    app = from_context(ctx)
+    if app.vault.delete(key):
+        app.print(f"Deleted key {key}")
+    else:
+        app.error(f"Key {key} not found")
+
+
+@vault.command()
+@click.argument("key")
+@click.pass_context
+def show(ctx: click.Context, key: str) -> None:
+    """Show the value of an API key"""
+    app = from_context(ctx)
+    if value := app.vault.get(key):
+        app.print(value)
+    else:
+        app.error(f"Key {key} not found")
+
+
+@vault.command()
+@click.pass_context
+def keys(ctx: click.Context) -> None:
+    """Show the keys in the vault"""
+    app = from_context(ctx)
+    for key in app.vault.keys():
+        app.print(key)

quantflow/cli/commands/vault.py

@@ -9,3 +9,4 @@
 SETTINGS_DIRECTORY = HOME_DIRECTORY / ".quantflow"
 SETTINGS_ENV_FILE = SETTINGS_DIRECTORY / ".env"
 HIST_FILE_PATH = SETTINGS_DIRECTORY / ".quantflow.his"
+VAULT_FILE_PATH = SETTINGS_DIRECTORY / ".vault"

quantflow/cli/settings.py

@@ -9,13 +9,19 @@ class Fred(AioHttpClient):
     url: str = "https://api.stlouisfed.org/fred"
     key: str = field(default_factory=lambda: os.environ.get("FRED_API_KEY", ""))
 
-    async def categiories(self, **kw: Any) -> list[dict]:
+    async def categiories(self, **kw: Any) -> dict:
         return await self.get_path("category", **kw)
 
+    async def subcategories(self, **kw: Any) -> dict:
+        return await self.get_path("category/children", **kw)
+
+    async def series(self, **kw: Any) -> dict:
+        return await self.get_path("category/series", **kw)
+
     # Internals
-    async def get_path(self, path: str, **kw: Any) -> list[dict]:
+    async def get_path(self, path: str, **kw: Any) -> dict:
         result = await self.get(f"{self.url}/{path}", **self.params(**kw))
-        return cast(list[dict], result)
+        return cast(dict, result)
 
     def params(self, params: dict | None = None, **kw: Any) -> dict:
         params = params.copy() if params is not None else {}

quantflow/data/fred.py

@@ -0,0 +1,39 @@
+from pathlib import Path
+
+
+class Vault:
+
+    def __init__(self, path: str | Path) -> None:
+        self.path = Path(path)
+        self.path.touch(exist_ok=True)
+        self.data = self.load()
+
+    def load(self) -> dict[str, str]:
+        data = {}
+        with open(self.path) as file:
+            for line in file:
+                key, value = line.strip().split("=")
+                data[key] = value
+        return data
+
+    def add(self, key: str, value: str) -> None:
+        self.data[key] = value
+        self.save()
+
+    def delete(self, key: str) -> bool:
+        if self.data.pop(key, None) is not None:
+            self.save()
+            return True
+        return False
+
+    def get(self, key: str) -> str | None:
+        return self.data.get(key)
+
+    def keys(self) -> list[str]:
+        return sorted(self.data)
+
+    def save(self) -> None:
+        with open(self.path, "w") as file:
+            for key in sorted(self.data):
+                value = self.data[key]
+                file.write(f"{key}={value}\n")