Fred charts (#12)

Author: lsbardel

dev/lint

@@ -1,14 +1,18 @@
 #!/usr/bin/env bash
 set -e
 
+ISORT_ARGS="-c"
 BLACK_ARG="--check"
 RUFF_ARG=""
 
 if [ "$1" = "fix" ] ; then
+    ISORT_ARGS=""
     BLACK_ARG=""
     RUFF_ARG="--fix"
 fi
 
+echo isort
+isort quantflow quantflow_tests ${ISORT_ARGS}
 echo black
 black quantflow quantflow_tests ${BLACK_ARG}
 echo ruff

poetry.lock

@@ -31,6 +31,7 @@ mypy = "^1.13.0"
 ghp-import = "^2.0.2"
 ruff = "^0.8.1"
 pytest-asyncio = "^0.25.0"
+isort = "^5.13.2"
 
 
 [tool.poetry.extras]

pyproject.toml

@@ -1,18 +1,20 @@
 import os
 from dataclasses import dataclass, field
-from typing import Any
 from functools import partial
+from typing import Any
+
 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 quantflow.data.vault import Vault
 
 from . import settings
+from .commands import fred, stocks, vault
 
 
 @click.group()

quantflow/cli/app.py

@@ -0,0 +1,43 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Self, cast
+
+import click
+
+from quantflow.data.fmp import FMP
+from quantflow.data.fred import Fred
+
+if TYPE_CHECKING:
+    from quantflow.cli.app import QfApp
+
+
+class QuantContext(click.Context):
+
+    @classmethod
+    def current(cls) -> Self:
+        return cast(Self, click.get_current_context())
+
+    @property
+    def qf(self) -> QfApp:
+        return self.obj  # type: ignore
+
+    def fmp(self) -> FMP:
+        if key := self.qf.vault.get("fmp"):
+            return FMP(key=key)
+        else:
+            raise click.UsageError("No FMP API key found")
+
+    def fred(self) -> Fred:
+        if key := self.qf.vault.get("fred"):
+            return Fred(key=key)
+        else:
+            raise click.UsageError("No FRED API key found")
+
+
+class QuantCommand(click.Command):
+    context_class = QuantContext
+
+
+class QuantGroup(click.Group):
+    context_class = QuantContext
+    command_class = QuantCommand

quantflow/cli/commands/base.py

@@ -1,68 +1,149 @@
 from __future__ import annotations
 
-import click
 import asyncio
-import pandas as pd
 from typing import TYPE_CHECKING
+
+import click
+import pandas as pd
+from asciichartpy import plot
+from ccy.cli.console import df_to_rich
 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
 
+from quantflow.data.fred import Fred
+
+from .base import QuantContext, QuantGroup
 
-FREQUENCIES = tuple(FMP().historical_frequencies())
+FREQUENCIES = tuple(Fred.freq)
 
 if TYPE_CHECKING:
-    from quantflow.cli.app import QfApp
+    pass
 
 
-@click.group(invoke_without_command=True)
-@click.pass_context
-def fred(ctx: click.Context) -> None:
+@click.group(invoke_without_command=True, cls=QuantGroup)
+def fred() -> None:
     """Federal Reserve of St. Louis data"""
+    ctx = QuantContext.current()
     if ctx.invoked_subcommand is None:
-        app = from_context(ctx)
-        app.print("Welcome to FRED data commands!")
-        app.print(ctx.get_help())
+        ctx.qf.print("Welcome to FRED data commands!")
+        ctx.qf.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:
+def subcategories(category_id: str | None = None) -> None:
     """List subcategories for a Fred category"""
-    app = from_context(ctx)
+    ctx = QuantContext.current()
     try:
-        data = asyncio.run(get_subcategories(app, category_id))
+        data = asyncio.run(get_subcategories(ctx, category_id))
     except HttpResponseError as e:
-        app.error(e)
+        ctx.qf.error(e)
     else:
         df = pd.DataFrame(data["categories"], columns=["id", "name"])
-        app.print(df_to_rich(df))
+        ctx.qf.print(df_to_rich(df))
 
 
 @fred.command()
-@click.pass_context
 @click.argument("category-id")
-def series(ctx: click.Context, category_id: str) -> None:
+def series(category_id: str) -> None:
     """List series for a Fred category"""
-    app = from_context(ctx)
+    ctx = QuantContext.current()
     try:
-        data = asyncio.run(get_series(app, category_id))
+        data = asyncio.run(get_series(ctx, category_id))
     except HttpResponseError as e:
-        app.error(e)
+        ctx.qf.error(e)
     else:
-        app.print(data)
+        ctx.qf.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:
+@fred.command()
+@click.argument("series-id")
+@click.option(
+    "-l",
+    "--length",
+    type=int,
+    default=100,
+    show_default=True,
+    help="Number of data points",
+)
+@click.option(
+    "-f",
+    "--frequency",
+    type=click.Choice(FREQUENCIES),
+    default="d",
+    show_default=True,
+    help="Frequency of data",
+)
+def data(series_id: str, length: int, frequency: str) -> None:
+    """Display a series data"""
+    ctx = QuantContext.current()
+    try:
+        df = asyncio.run(get_serie_data(ctx, series_id, length, frequency))
+    except HttpResponseError as e:
+        ctx.qf.error(e)
+    else:
+        ctx.qf.print(df_to_rich(df))
+
+
+@fred.command()
+@click.argument("series-id")
+@click.option(
+    "-h",
+    "--height",
+    type=int,
+    default=20,
+    show_default=True,
+    help="Chart height",
+)
+@click.option(
+    "-l",
+    "--length",
+    type=int,
+    default=100,
+    show_default=True,
+    help="Number of data points",
+)
+@click.option(
+    "-f",
+    "--frequency",
+    type=click.Choice(FREQUENCIES),
+    default="w",
+    show_default=True,
+    help="Frequency of data",
+)
+def chart(series_id: str, height: int, length: int, frequency: str) -> None:
+    """Chart a serie"""
+    ctx = QuantContext.current()
+    try:
+        df = asyncio.run(get_serie_data(ctx, series_id, length, frequency))
+    except HttpResponseError as e:
+        ctx.qf.error(e)
+    else:
+        data = list(reversed(df["value"].tolist()[:length]))
+        ctx.qf.print(plot(data, {"height": height}))
+
+
+async def get_subcategories(ctx: QuantContext, category_id: str | None) -> dict:
+    async with ctx.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:
+async def get_series(ctx: QuantContext, category_id: str) -> dict:
+    async with ctx.fred() as cli:
         return await cli.series(params=compact_dict(category_id=category_id))
+
+
+async def get_serie_data(
+    ctx: QuantContext, series_id: str, length: int, frequency: str
+) -> dict:
+    async with ctx.fred() as cli:
+        return await cli.serie_data(
+            params=dict(
+                series_id=series_id,
+                limit=length,
+                frequency=frequency,
+                sort_order="desc",
+            )
+        )