Working on client (#7)

Author: lsbardel

.gitignore

@@ -27,6 +27,7 @@ dist
 .venv
 .mypy_cache
 .pytest_cache
+.ruff_cache
 .python-version
 
 # Jupyter

poetry.lock

@@ -15,13 +15,14 @@ Documentation = "https://quantmind.github.io/quantflow/"
 python = ">=3.11"
 scipy = "^1.14.1"
 pydantic = "^2.0.2"
-ccy = {version="1.6.0", extras=["cli"]}
-asciichart = "^0.1"
+ccy = {version="1.6.0"}
 python-dotenv = "^1.0.1"
-asciichartpy = "^1.5.25"
-prompt-toolkit = "^3.0.43"
 polars = {version = "^1.11.0", extras=["pandas", "pyarrow"]}
+asciichartpy = { version = "^1.5.25", optional = true }
+prompt-toolkit = { version = "^3.0.43", optional = true }
 aio-fluid = {version = "^1.2.1", extras=["http"], optional = true}
+rich = {version = "^13.9.4", optional = true}
+click = {version = "^8.1.7", optional = true}
 
 [tool.poetry.group.dev.dependencies]
 black = "^24.1.1"
@@ -34,6 +35,7 @@ pytest-asyncio = "^0.24.0"
 
 [tool.poetry.extras]
 data = ["aio-fluid"]
+cli = ["asciichartpy", "prompt-toolkit", "rich", "click"]
 
 [tool.poetry.group.book]
 optional = true
@@ -48,7 +50,7 @@ sympy = "^1.12"
 ipywidgets = "^8.0.7"
 
 [tool.poetry.scripts]
-qf = "quantflow.cli:main"
+qf = "quantflow.cli.script:main"
 
 [build-system]
 requires = ["poetry-core>=1.0.0"]
@@ -68,7 +70,6 @@ profile = "black"
 
 [tool.ruff]
 lint.select = ["E", "F"]
-extend-exclude = ["fluid_apps/db/migrations"]
 line-length = 88
 
 [tool.hatch.version]

pyproject.toml

@@ -1,131 +0,0 @@
-import asyncio
-import os
-from dataclasses import dataclass, field
-from typing import Any
-
-import click
-import dotenv
-import pandas as pd
-from asciichartpy import plot
-from ccy.cli.console import df_to_rich
-from prompt_toolkit import PromptSession
-from prompt_toolkit.history import FileHistory
-from rich.console import Console
-from rich.text import Text
-
-from quantflow.data.fmp import FMP
-
-from . import settings
-
-dotenv.load_dotenv()
-
-FREQUENCIES = tuple(FMP().historical_frequencies())
-
-
-@click.group()
-def qf() -> None:
-    pass
-
-
-@qf.command()
-@click.argument("symbol")
-def profile(symbol: str) -> None:
-    """Company profile"""
-    data = asyncio.run(get_profile(symbol))[0]
-    main.print(data.pop("description"))
-    df = pd.DataFrame(data.items(), columns=["Key", "Value"])
-    main.print(df_to_rich(df))
-
-
-@qf.command()
-@click.argument("symbol")
-@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="",
-    help="Number of data points",
-)
-def chart(symbol: str, height: int, length: int, frequency: str) -> None:
-    """Symbol chart"""
-    df = asyncio.run(get_prices(symbol, frequency))
-    data = list(reversed(df["close"].tolist()[:length]))
-    print(plot(data, {"height": height}))
-
-
-async def get_prices(symbol: str, frequency: str) -> pd.DataFrame:
-    async with FMP() as cli:
-        return await cli.prices(symbol, frequency)
-
-
-async def get_profile(symbol: str) -> list[dict]:
-    async with FMP() as cli:
-        return await cli.profile(symbol)
-
-
-@dataclass
-class App:
-    console: Console = field(default_factory=Console)
-
-    def __call__(self) -> None:
-        os.makedirs(settings.SETTINGS_DIRECTORY, exist_ok=True)
-        history = FileHistory(str(settings.HIST_FILE_PATH))
-        session: PromptSession = PromptSession(history=history)
-
-        self.print("Welcome to QuantFlow!", style="bold green")
-        self.handle_command("help")
-
-        try:
-            while True:
-                try:
-                    text = session.prompt("quantflow> ")
-                except KeyboardInterrupt:
-                    break
-                else:
-                    self.handle_command(text)
-        except click.Abort:
-            self.console.print(Text("Bye!", style="bold magenta"))
-
-    def print(self, text_alike: Any, style: str = "") -> None:
-        if isinstance(text_alike, str):
-            style = style or "cyan"
-            text_alike = Text(f"\n{text_alike}\n", style="cyan")
-        self.console.print(text_alike)
-
-    def error(self, err: str | Exception) -> None:
-        self.console.print(Text(f"\n{err}\n", style="bold red"))
-
-    def handle_command(self, text: str) -> None:
-        self.current_command = text.split(" ")[0].strip()
-        if not text:
-            return
-        elif text == "help":
-            return qf.main(["--help"], standalone_mode=False)
-        elif text == "exit":
-            raise click.Abort()
-
-        try:
-            qf.main(text.split(), standalone_mode=False)
-        except click.exceptions.MissingParameter as e:
-            self.error(e)
-        except click.exceptions.NoSuchOption as e:
-            self.error(e)
-
-
-main = App()

quantflow/cli/__init__.py

@@ -0,0 +1,134 @@
+import asyncio
+import os
+from dataclasses import dataclass, field
+from typing import Any, Self
+
+import click
+import dotenv
+import pandas as pd
+from asciichartpy import plot
+from ccy.cli.console import df_to_rich
+from prompt_toolkit import PromptSession
+from prompt_toolkit.history import FileHistory
+from rich.console import Console
+from rich.text import Text
+
+from quantflow.data.fmp import FMP
+
+from . import settings
+
+dotenv.load_dotenv()
+
+FREQUENCIES = tuple(FMP().historical_frequencies())
+
+
+@click.group()
+def qf() -> None:
+    pass
+
+
+@qf.command()
+@click.argument("symbol")
+@click.pass_context
+def profile(ctx: click.Context, symbol: str) -> None:
+    """Company profile"""
+    app = QfApp.from_context(ctx)
+    data = asyncio.run(get_profile(symbol))[0]
+    app.print(data.pop("description"))
+    df = pd.DataFrame(data.items(), columns=["Key", "Value"])
+    app.print(df_to_rich(df))
+
+
+@qf.command()
+@click.argument("symbol")
+@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="",
+    help="Number of data points",
+)
+def chart(symbol: str, height: int, length: int, frequency: str) -> None:
+    """Symbol chart"""
+    df = asyncio.run(get_prices(symbol, frequency))
+    data = list(reversed(df["close"].tolist()[:length]))
+    print(plot(data, {"height": height}))
+
+
+async def get_prices(symbol: str, frequency: str) -> pd.DataFrame:
+    async with FMP() as cli:
+        return await cli.prices(symbol, frequency)
+
+
+async def get_profile(symbol: str) -> list[dict]:
+    async with FMP() as cli:
+        return await cli.profile(symbol)
+
+
+@dataclass
+class QfApp:
+    console: Console = field(default_factory=Console)
+
+    @classmethod
+    def from_context(cls, ctx: click.Context) -> Self:
+        return ctx.obj  # type: ignore
+
+    def __call__(self) -> None:
+        os.makedirs(settings.SETTINGS_DIRECTORY, exist_ok=True)
+        history = FileHistory(str(settings.HIST_FILE_PATH))
+        session: PromptSession = PromptSession(history=history)
+
+        self.print("Welcome to QuantFlow!", style="bold green")
+        self.handle_command("help")
+
+        try:
+            while True:
+                try:
+                    text = session.prompt("quantflow> ")
+                except KeyboardInterrupt:
+                    break
+                else:
+                    self.handle_command(text)
+        except click.Abort:
+            self.console.print(Text("Bye!", style="bold magenta"))
+
+    def print(self, text_alike: Any, style: str = "") -> None:
+        if isinstance(text_alike, str):
+            style = style or "cyan"
+            text_alike = Text(f"\n{text_alike}\n", style="cyan")
+        self.console.print(text_alike)
+
+    def error(self, err: str | Exception) -> None:
+        self.console.print(Text(f"\n{err}\n", style="bold red"))
+
+    def handle_command(self, text: str) -> None:
+        self.current_command = text.split(" ")[0].strip()
+        if not text:
+            return
+        elif text == "help":
+            return qf.main(["--help"], standalone_mode=False, obj=self)
+        elif text == "exit":
+            raise click.Abort()
+
+        try:
+            qf.main(text.split(), standalone_mode=False, obj=self)
+        except click.exceptions.MissingParameter as e:
+            self.error(e)
+        except click.exceptions.NoSuchOption as e:
+            self.error(e)

quantflow/cli/app.py

@@ -0,0 +1,10 @@
+try:
+    from .app import QfApp
+except ImportError:
+    raise ImportError(
+        "Cannot run qf command line, "
+        "quantflow needs to be installed with cli & data extras, "
+        "pip install quantflow[cli, data]"
+    ) from None
+
+main = QfApp()

quantflow/cli/script.py

@@ -28,5 +28,10 @@ pip install quantflow
 
 ## Command line tools
 
-When installing with the extra `data` dependencies, it is possible to use the command line tool `qf`
+The command line tools are available when installing with the extra `cli` and `data` dependencies.
 
+```bash
+pip install quantflow[cli,data]
+```
+
+It is possible to use the command line tool `qf` to download data and run pricing and calibration scripts.