Add options and prices for crypto

Author: lsbardel

notebooks/api/options/vol_surface.rst

@@ -29,3 +29,7 @@ Vol Surface
    :member-order: groupwise
    :autosummary:
    :autosummary-nosignatures:
+
+
+.. autoclass:: OptionSelection
+   :members:

notebooks/index.md

@@ -24,8 +24,25 @@ This documentation is organized into a few major sections.
 * **Stochastic Processes** cover all the stochastic models supported and their use
 * **Applications** show case the real-world use cases
 * **Examples** random examples
-* **API Reference** python API reference``
+* **API Reference** python API reference
 
-```{code-cell} ipython3
+## Installation
 
+To install the library use
 ```
+pip install quantflow
+```
+
+
+## Optional dependencies
+
+Quantflow comes with two optional dependencies:
+
+* `data` for data retrieval, to install it use
+  ```
+  pip install quantflow[data]
+  ```
+* `cli` for command line interface, to install it use
+  ```
+  pip install quantflow[data,cli]
+  ```

quantflow/cli/app.py

@@ -6,6 +6,7 @@
 import click
 from prompt_toolkit import PromptSession
 from prompt_toolkit.completion import NestedCompleter
+from prompt_toolkit.formatted_text import HTML
 from prompt_toolkit.history import FileHistory
 from rich.console import Console
 from rich.text import Text
@@ -38,6 +39,7 @@ def __call__(self) -> None:
                         self.prompt_message(),
                         completer=self.prompt_completer(),
                         complete_while_typing=True,
+                        bottom_toolbar=self.bottom_toolbar,
                     )
                 except KeyboardInterrupt:
                     break
@@ -82,3 +84,15 @@ def handle_command(self, text: str) -> None:
             click.exceptions.UsageError,
         ) as e:
             self.error(e)
+
+    def bottom_toolbar(self) -> HTML:
+        sections = "/".join([str(section.name) for section in self.sections])
+        back = (
+            (' <b><style bg="ansired">back</style></b> ' "to exit the current section,")
+            if len(self.sections) > 1
+            else ""
+        )
+        return HTML(
+            f"Your are in <strong>{sections}</strong>, type{back} "
+            '<b><style bg="ansired">exit</style></b> to exit'
+        )

quantflow/cli/commands/base.py

@@ -12,6 +12,9 @@
     from quantflow.cli.app import QfApp
 
 
+FREQUENCIES = tuple(FMP().historical_frequencies())
+
+
 class HistoricalPeriod(enum.StrEnum):
     day = "1d"
     week = "1w"
@@ -122,3 +125,10 @@ class options:
         default=-1,
         help="maturity index",
     )
+    frequency = click.option(
+        "-f",
+        "--frequency",
+        type=click.Choice(FREQUENCIES),
+        default="",
+        help="Frequency of data - if not provided it is daily",
+    )

quantflow/cli/commands/crypto.py

@@ -10,8 +10,10 @@
 
 from quantflow.data.deribit import Deribit
 from quantflow.options.surface import VolSurface
+from quantflow.utils.numbers import round_to_step
 
 from .base import QuantContext, options, quant_group
+from .stocks import get_prices
 
 
 @quant_group()
@@ -64,15 +66,42 @@ def implied_vol(currency: str, index: int, height: int, chart: bool) -> None:
     index_or_none = None if index < 0 else index
     vs.bs(index=index_or_none)
     df = vs.options_df(index=index_or_none)
-    df["implied_vol"] = df["implied_vol"] * 100
-    df = df.round({"ttm": 4, "moneyness": 4, "moneyness_ttm": 4, "implied_vol": 5})
     if chart:
-        data = df["implied_vol"].tolist()
+        data = (df["implied_vol"] * 100).tolist()
         ctx.qf.print(plot(data, {"height": height}))
     else:
+        df[["ttm", "moneyness", "moneyness_ttm"]] = df[
+            ["ttm", "moneyness", "moneyness_ttm"]
+        ].map("{:.4f}".format)
+        df["implied_vol"] = df["implied_vol"].map("{:.2%}".format)
+        df["price"] = df["price"].map(lambda p: round_to_step(p, vs.tick_size_options))
+        df["forward_price"] = df["forward_price"].map(
+            lambda p: round_to_step(p, vs.tick_size_forwards)
+        )
         ctx.qf.print(df_to_rich(df))
 
 
+@crypto.command()
+@click.argument("symbol")
+@options.height
+@options.length
+@options.chart
+@options.frequency
+def prices(symbol: str, height: int, length: int, chart: bool, frequency: str) -> None:
+    """Fetch OHLC prices for given cryptocurrency"""
+    ctx = QuantContext.current()
+    df = asyncio.run(get_prices(ctx, symbol, frequency))
+    if df.empty:
+        raise click.UsageError(
+            f"No data for {symbol} - are you sure the symbol exists?"
+        )
+    if chart:
+        data = list(reversed(df["close"].tolist()[:length]))
+        ctx.qf.print(plot(data, {"height": height}))
+    else:
+        ctx.qf.print(df_to_rich(df[["date", "open", "high", "low", "close", "volume"]]))
+
+
 async def get_volatility(ctx: QuantContext, currency: str) -> pd.DataFrame:
     async with Deribit() as client:
         return await client.get_volatility(params=dict(currency=currency))

quantflow/cli/commands/stocks.py

@@ -11,13 +11,10 @@
 from ccy.cli.console import df_to_rich
 from ccy.tradingcentres import prevbizday
 
-from quantflow.data.fmp import FMP
 from quantflow.utils.dates import utcnow
 
 from .base import HistoricalPeriod, QuantContext, options, quant_group
 
-FREQUENCIES = tuple(FMP().historical_frequencies())
-
 
 @quant_group()
 def stocks() -> None:
@@ -56,13 +53,7 @@ def search(text: str) -> None:
 @click.argument("symbol")
 @options.height
 @options.length
-@click.option(
-    "-f",
-    "--frequency",
-    type=click.Choice(FREQUENCIES),
-    default="",
-    help="Frequency of data - if not provided it is daily",
-)
+@options.frequency
 def chart(symbol: str, height: int, length: int, frequency: str) -> None:
     """Symbol chart"""
     ctx = QuantContext.current()

quantflow/data/deribit.py

@@ -1,12 +1,13 @@
 from datetime import datetime, timezone
+from decimal import Decimal
 from typing import Any, cast
 
 import pandas as pd
 from dateutil.parser import parse
 from fluid.utils.http_client import AioHttpClient, HttpResponse
 
 from quantflow.options.surface import VolSecurityType, VolSurfaceLoader
-from quantflow.utils.numbers import round_to_step
+from quantflow.utils.numbers import round_to_step, to_decimal
 
 
 def parse_maturity(v: str) -> datetime:
@@ -55,12 +56,13 @@ async def volatility_surface_loader(self, currency: str) -> VolSurfaceLoader:
         )
         instruments = await self.get_instruments(params=dict(currency=currency))
         instrument_map = {i["instrument_name"]: i for i in instruments}
-
+        min_tick_size = Decimal("inf")
         for future in futures:
             if (bid_ := future["bid_price"]) and (ask_ := future["ask_price"]):
                 name = future["instrument_name"]
                 meta = instrument_map[name]
-                tick_size = meta["tick_size"]
+                tick_size = to_decimal(meta["tick_size"])
+                min_tick_size = min(min_tick_size, tick_size)
                 bid = round_to_step(bid_, tick_size)
                 ask = round_to_step(ask_, tick_size)
                 if meta["settlement_period"] == "perpetual":
@@ -85,12 +87,15 @@ async def volatility_surface_loader(self, currency: str) -> VolSurfaceLoader:
                         open_interest=int(future["open_interest"]),
                         volume=int(future["volume_usd"]),
                     )
+        loader.tick_size_forwards = min_tick_size
 
+        min_tick_size = Decimal("inf")
         for option in options:
             if (bid_ := option["bid_price"]) and (ask_ := option["ask_price"]):
                 name = option["instrument_name"]
                 meta = instrument_map[name]
-                tick_size = meta["tick_size"]
+                tick_size = to_decimal(meta["tick_size"])
+                min_tick_size = min(min_tick_size, tick_size)
                 loader.add_option(
                     VolSecurityType.option,
                     strike=round_to_step(meta["strike"], tick_size),
@@ -103,7 +108,7 @@ async def volatility_surface_loader(self, currency: str) -> VolSurfaceLoader:
                     bid=round_to_step(bid_, tick_size),
                     ask=round_to_step(ask_, tick_size),
                 )
-
+        loader.tick_size_options = min_tick_size
         return loader
 
     # Internal methods

quantflow/options/surface.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import enum
+import warnings
 from dataclasses import dataclass, field
 from datetime import datetime, timedelta
 from decimal import Decimal
@@ -39,14 +40,21 @@ def vol_surface_type(self) -> VolSecurityType: ...
 
 
 class OptionSelection(enum.Enum):
-    """Option selection method"""
+    """Option selection method
+
+    This enum is used to select which one between calls and puts are used
+    for calculating implied volatility and other operations
+    """
 
     best = enum.auto()
-    """Select the best bid/ask options - the ones which are otm"""
+    """Select the best bid/ask options.
+
+    These are the options which are Out of the Money, where their
+    intrinsic value is zero"""
     call = enum.auto()
-    """Select the call options"""
+    """Select the call options only"""
     put = enum.auto()
-    """Select the put options"""
+    """Select the put options only"""
 
 
 @dataclass
@@ -179,6 +187,10 @@ def put_price(self) -> Decimal:
         else:
             return self.price
 
+    @property
+    def option_type(self) -> str:
+        return "call" if self.call else "put"
+
     def can_price(self, converged: bool, select: OptionSelection) -> bool:
         if self.price_time > ZERO and not np.isnan(self.implied_vol):
             if not self.converged and converged is True:
@@ -221,7 +233,7 @@ def _asdict(self) -> dict[str, Any]:
             price=float(self.price),
             price_bp=float(self.price_bp),
             forward_price=float(self.forward_price),
-            call=self.call,
+            type=self.option_type,
             side=self.side,
         )
 
@@ -396,6 +408,10 @@ class VolSurface(Generic[S]):
     """Sorted tuple of :class:`.VolCrossSection` for different maturities"""
     day_counter: DayCounter = default_day_counter
     """Day counter for time to maturity calculations - by default it uses Act/Act"""
+    tick_size_forwards: Decimal | None = None
+    """Tick size for rounding forward and spot prices - optional"""
+    tick_size_options: Decimal | None = None
+    """Tick size for rounding option prices - optional"""
 
     def securities(self) -> Iterator[SpotPrice[S] | FwdPrice[S] | OptionPrices[S]]:
         """Iterator over all securities in the volatility surface"""
@@ -457,17 +473,30 @@ def bs(
         initial_vol: float = INITIAL_VOL,
     ) -> list[OptionPrice]:
         """calculate Black-Scholes implied volatility for all options
-        in the surface"""
+        in the surface
+
+        :param select: the :class:`.OptionSelection` method
+        :param index: Index of the cross section to use, if None use all
+        :param initial_vol: Initial volatility for the root finding algorithm
+
+        Some options may not converge, in this case the implied volatility is not
+        calculated correctly and the option is marked as not converged.
+        """
         d = self.as_array(
-            select=select, index=index, initial_vol=initial_vol, converged=False
-        )
-        result = implied_black_volatility(
-            k=d.moneyness,
-            price=d.price,
-            ttm=d.ttm,
-            initial_sigma=d.implied_vol,
-            call_put=d.call_put,
+            select=select,
+            index=index,
+            initial_vol=initial_vol,
+            converged=False,
         )
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore")
+            result = implied_black_volatility(
+                k=d.moneyness,
+                price=d.price,
+                ttm=d.ttm,
+                initial_sigma=d.implied_vol,
+                call_put=d.call_put,
+            )
         for option, implied_vol, converged in zip(
             d.options, result.root, result.converged
         ):
@@ -495,7 +524,10 @@ def options_df(
     ) -> pd.DataFrame:
         """Time frame of Black-Scholes call input data"""
         data = self.option_prices(
-            select=select, index=index, initial_vol=initial_vol, converged=converged
+            select=select,
+            index=index,
+            initial_vol=initial_vol,
+            converged=converged,
         )
         return pd.DataFrame([d._asdict() for d in data])
 
@@ -639,8 +671,15 @@ class GenericVolSurfaceLoader(Generic[S]):
     """Helper class to build a volatility surface from a list of securities"""
 
     spot: SpotPrice[S] | None = None
+    """Spot price of the underlying asset"""
     maturities: dict[datetime, VolCrossSectionLoader[S]] = field(default_factory=dict)
+    """Dictionary of maturities and their corresponding cross section loaders"""
     day_counter: DayCounter = default_day_counter
+    """Day counter for time to maturity calculations - by default it uses Act/Act"""
+    tick_size_forwards: Decimal | None = None
+    """Tick size for rounding forward and spot prices - optional"""
+    tick_size_options: Decimal | None = None
+    """Tick size for rounding option prices - optional"""
 
     def get_or_create_maturity(self, maturity: datetime) -> VolCrossSectionLoader[S]:
         if maturity not in self.maturities:
@@ -727,6 +766,8 @@ def surface(self, ref_date: datetime | None = None) -> VolSurface[S]:
             spot=self.spot,
             maturities=tuple(maturities),
             day_counter=self.day_counter,
+            tick_size_forwards=self.tick_size_forwards,
+            tick_size_options=self.tick_size_options,
         )