Command-Line Interface (CLI)

This CLI package contains the main entry point and command structure for the optpricing command-line interface.

backtest

backtest(
    ticker: Annotated[
        str,
        Option(
            "--ticker",
            "-t",
            help="The stock ticker to backtest.",
        ),
    ],
    model: Annotated[
        str,
        Option(
            "--model",
            "-m",
            help="The single model to backtest.",
        ),
    ],
    verbose: Annotated[
        bool,
        Option(
            "--verbose",
            "-v",
            help="Enable detailed logging.",
        ),
    ] = False,
)

Runs a historical backtest for a given model and ticker.

Source code in src/optpricing/cli/commands/backtest.py

def backtest(
    ticker: Annotated[
        str, typer.Option("--ticker", "-t", help="The stock ticker to backtest.")
    ],
    model: Annotated[
        str, typer.Option("--model", "-m", help="The single model to backtest.")
    ],
    verbose: Annotated[
        bool, typer.Option("--verbose", "-v", help="Enable detailed logging.")
    ] = False,
):
    """
    Runs a historical backtest for a given model and ticker.
    """
    from optpricing.cli.main import setup_logging

    setup_logging(verbose)

    if model not in ALL_MODEL_CONFIGS:
        typer.secho(
            f"Error: Model '{model}' not found. "
            f"Available: {list(ALL_MODEL_CONFIGS.keys())}",
            fg=typer.colors.RED,
        )
        raise typer.Exit(code=1)

    config = ALL_MODEL_CONFIGS[model].copy()
    config["ticker"] = ticker

    workflow = BacktestWorkflow(ticker, config)
    workflow.run()
    workflow.save_results()

calibrate

calibrate(
    ticker: Annotated[
        str,
        Option(
            "--ticker",
            "-t",
            help="The stock ticker to calibrate against.",
        ),
    ],
    model: Annotated[
        list[str],
        Option(
            "--model",
            "-m",
            help="Model to calibrate. Can be used multiple times.",
        ),
    ],
    date: Annotated[
        str | None,
        Option(
            "--date",
            "-d",
            help="Snapshot date (YYYY-MM-DD). Defaults to latest available.",
        ),
    ] = None,
    fix_param: Annotated[
        list[str] | None,
        Option(
            "--fix",
            help="Fix a parameter (e.g., 'sigma=0.25'). Can be used multiple times.",
        ),
    ] = None,
    verbose: Annotated[
        bool,
        Option(
            "--verbose",
            "-v",
            help="Enable detailed logging.",
        ),
    ] = False,
)

Calibrates one or more models to market data for a given ticker and date.

Source code in src/optpricing/cli/commands/calibrate.py

def calibrate(
    ticker: Annotated[
        str,
        typer.Option("--ticker", "-t", help="The stock ticker to calibrate against."),
    ],
    model: Annotated[
        list[str],
        typer.Option(
            "--model", "-m", help="Model to calibrate. Can be used multiple times."
        ),
    ],
    date: Annotated[
        str | None,
        typer.Option(
            "--date",
            "-d",
            help="Snapshot date (YYYY-MM-DD). Defaults to latest available.",
        ),
    ] = None,
    fix_param: Annotated[
        list[str] | None,
        typer.Option(
            "--fix",
            help="Fix a parameter (e.g., 'sigma=0.25'). Can be used multiple times.",
        ),
    ] = None,
    verbose: Annotated[
        bool, typer.Option("--verbose", "-v", help="Enable detailed logging.")
    ] = False,
):
    """
    Calibrates one or more models to market data for a given ticker and date.
    """
    from optpricing.cli.main import setup_logging

    setup_logging(verbose)

    current_dir = Path.cwd()
    artifacts_base_dir = current_dir / _config.get("artifacts_directory", "artifacts")
    calibrated_params_dir = artifacts_base_dir / "calibrated_params"
    calibrated_params_dir.mkdir(parents=True, exist_ok=True)

    if date is None:
        typer.echo(
            f"No date specified for {ticker}. Finding latest available snapshot..."
        )
        available_dates = get_available_snapshot_dates(ticker)
        if not available_dates:
            typer.secho(
                f"Error: No market data snapshots found for ticker '{ticker}'.",
                fg=typer.colors.RED,
            )
            raise typer.Exit(code=1)
        date = available_dates[0]
        typer.echo(f"Using latest date: {date}")

    market_data = load_market_snapshot(ticker, date)
    if market_data is None:
        typer.secho(
            f"Error: Failed to load market data for {ticker} on {date}.",
            fg=typer.colors.RED,
        )
        raise typer.Exit(code=1)

    frozen_params = {}
    if fix_param:
        for p in fix_param:
            try:
                key, value = p.split("=")
                frozen_params[key.strip()] = float(value)
            except ValueError:
                typer.secho(
                    f"Invalid format for fixed parameter: '{p}'. Use 'key=value'.",
                    fg=typer.colors.RED,
                )
                raise typer.Exit(code=1)

    for model_name in model:
        if model_name not in AVAILABLE_MODELS_FOR_CALIBRATION:
            typer.secho(
                f"Warning: Model '{model_name}' is not supported for calibration. "
                f"Available: {list(AVAILABLE_MODELS_FOR_CALIBRATION.keys())}. "
                f"Skipping.",
                fg=typer.colors.YELLOW,
            )
            continue

        config = AVAILABLE_MODELS_FOR_CALIBRATION[model_name].copy()
        config["ticker"] = ticker
        config["frozen"] = {**config.get("frozen", {}), **frozen_params}

        workflow = DailyWorkflow(market_data, config)
        workflow.run()

        if workflow.results["Status"] == "Success":
            typer.secho(
                f"\nCalibration for {model_name} on {date} SUCCEEDED.",
                fg=typer.colors.GREEN,
            )
            typer.echo(f"  - Final RMSE: {workflow.results['RMSE']:.6f}")
            typer.echo(
                f"  - Calibrated Params: {workflow.results['Calibrated Params']}"
            )

            params_to_save = {
                "model": model_name,
                "ticker": ticker,
                "date": date,
                "params": workflow.results["Calibrated Params"],
            }
            filename = f"{ticker}_{model_name}_{date}.json"
            save_path = calibrated_params_dir / filename
            with open(save_path, "w") as f:
                json.dump(params_to_save, f, indent=4)
            typer.echo(f"  - Saved parameters to: {save_path}")
        else:
            typer.secho(
                f"\nCalibration for {model_name} on {date} FAILED.",
                fg=typer.colors.RED,
            )
            typer.echo(f"  - Error: {workflow.results.get('Error', 'Unknown error')}")

dashboard

dashboard()

Launches the Streamlit dashboard application.

Source code in src/optpricing/cli/commands/dashboard.py

def dashboard():
    """
    Launches the Streamlit dashboard application.
    """
    try:
        with resources.path("optpricing.dashboard", "Home.py") as app_path:
            typer.echo(f"Launching Streamlit dashboard from: {app_path}")
            subprocess.run(["streamlit", "run", str(app_path)], check=True)
    except FileNotFoundError:
        typer.secho(
            "Error: 'streamlit' command not found. Hint: `pip install optpricing[app]`",
            fg=typer.colors.RED,
        )
        raise typer.Exit(code=1)

download_data

download_data(
    tickers: Annotated[
        list[str] | None,
        Option(
            "--ticker",
            "-t",
            help="Stock ticker to download. Can be used multiple times.",
        ),
    ] = None,
    all_default: Annotated[
        bool,
        Option(
            "--all",
            help="Download all default tickers specified in config.yaml.",
        ),
    ] = False,
    period: Annotated[
        str,
        Option(
            "--period",
            "-p",
            help="Time period for historical data (e.g., '10y', '5y').",
        ),
    ] = "10y",
)

Downloads and saves historical log returns for specified tickers or all defaults.

Source code in src/optpricing/cli/commands/data.py

def download_data(
    tickers: Annotated[
        list[str] | None,
        typer.Option(
            "--ticker",
            "-t",
            help="Stock ticker to download. Can be used multiple times.",
        ),
    ] = None,
    all_default: Annotated[
        bool,
        typer.Option(
            "--all", help="Download all default tickers specified in config.yaml."
        ),
    ] = False,
    period: Annotated[
        str,
        typer.Option(
            "--period",
            "-p",
            help="Time period for historical data (e.g., '10y', '5y').",
        ),
    ] = "10y",
):
    """
    Downloads and saves historical log returns for specified tickers or all defaults.
    """
    if all_default:
        tickers_to_process = _config.get("default_tickers", [])
        if not tickers_to_process:
            typer.secho(
                "Error: --all flag used, but no 'default_tickers' found "
                "in config.yaml.",
                fg=typer.colors.RED,
            )
            raise typer.Exit(code=1)
        typer.echo(f"Downloading all default tickers for period {period}...")
    elif tickers:
        tickers_to_process = tickers
        typer.echo(
            f"Downloading {period} historical data for tickers: "
            f"{', '.join(tickers_to_process)}"
        )
    else:
        typer.secho(
            "Error: Please provide at least one --ticker or use the --all flag.",
            fg=typer.colors.RED,
        )
        raise typer.Exit(code=1)

    save_historical_returns(tickers_to_process, period=period)
    typer.secho("Download complete.", fg=typer.colors.GREEN)

get_dividends

get_dividends(
    tickers: Annotated[
        list[str] | None,
        Option(
            "--ticker",
            "-t",
            help="Stock ticker to fetch. Can be used multiple times.",
        ),
    ] = None,
    all_default: Annotated[
        bool,
        Option(
            "--all",
            help="Fetch for all default tickers specified in config.yaml.",
        ),
    ] = False,
)

Fetches and displays the live forward dividend yield for specified tickers.

Source code in src/optpricing/cli/commands/data.py

def get_dividends(
    tickers: Annotated[
        list[str] | None,
        typer.Option(
            "--ticker",
            "-t",
            help="Stock ticker to fetch. Can be used multiple times.",
        ),
    ] = None,
    all_default: Annotated[
        bool,
        typer.Option(
            "--all", help="Fetch for all default tickers specified in config.yaml."
        ),
    ] = False,
):
    """
    Fetches and displays the live forward dividend yield for specified tickers.
    """
    if all_default:
        tickers_to_fetch = _config.get("default_tickers", [])
        if not tickers_to_fetch:
            typer.secho(
                "Error: --all flag used, but no 'default_tickers' in config.yaml.",
                fg=typer.colors.RED,
            )
            raise typer.Exit(code=1)
        typer.echo("Fetching dividend yields for all default tickers...")
    elif tickers:
        tickers_to_fetch = tickers
    else:
        typer.secho(
            "Error: Please provide at least one --ticker or use the --all flag.",
            fg=typer.colors.RED,
        )
        raise typer.Exit(code=1)

    console = Console()
    table = Table(title="Live Dividend Yields")
    table.add_column("Ticker", justify="left", style="cyan", no_wrap=True)
    table.add_column("Dividend Yield", justify="right", style="magenta")

    for ticker in tickers_to_fetch:
        yield_val = get_live_dividend_yield(ticker)
        table.add_row(ticker.upper(), f"{yield_val:.4%}")

    console.print(table)

save_snapshot

save_snapshot(
    tickers: Annotated[
        list[str] | None,
        Option(
            "--ticker",
            "-t",
            help="Stock ticker to snapshot. Can be used multiple times.",
        ),
    ] = None,
    all_default: Annotated[
        bool,
        Option(
            "--all",
            help="Snapshot all default tickers specified in config.yaml.",
        ),
    ] = False,
)

Fetches and saves a live market data snapshot for specified tickers.

Source code in src/optpricing/cli/commands/data.py

def save_snapshot(
    tickers: Annotated[
        list[str] | None,
        typer.Option(
            "--ticker",
            "-t",
            help="Stock ticker to snapshot. Can be used multiple times.",
        ),
    ] = None,
    all_default: Annotated[
        bool,
        typer.Option(
            "--all", help="Snapshot all default tickers specified in config.yaml."
        ),
    ] = False,
):
    """
    Fetches and saves a live market data snapshot for specified tickers.
    """
    if all_default:
        tickers_to_process = _config.get("default_tickers", [])
        if not tickers_to_process:
            typer.secho(
                "Error: --all flag used, but no 'default_tickers' in config.yaml.",
                fg=typer.colors.RED,
            )
            raise typer.Exit(code=1)
        typer.echo("Saving live market snapshots for all default tickers...")
    elif tickers:
        tickers_to_process = tickers
        typer.echo(
            f"Saving live market snapshots; tickers: {', '.join(tickers_to_process)}"
        )
    else:
        typer.secho(
            "Error: Please provide at least one --ticker or use the --all flag.",
            fg=typer.colors.RED,
        )
        raise typer.Exit(code=1)

    save_market_snapshot(tickers_to_process)
    typer.secho("Snapshot complete.", fg=typer.colors.GREEN)

demo_american

demo_american()

Runs the American options pricing benchmark.

Source code in src/optpricing/cli/commands/demo.py

@app.command(name="american")
def demo_american():
    """Runs the American options pricing benchmark."""
    _run_example("american_options_benchmark.py", [])

demo_european

demo_european(
    model: Annotated[
        str | None,
        Option(
            "--model",
            "-m",
            help="Run for a specific model (e.g., 'BSM').",
            case_sensitive=False,
        ),
    ] = None,
    technique: Annotated[
        str | None,
        Option(
            "--technique",
            "-t",
            help="Run for a specific technique (e.g., 'MC').",
            case_sensitive=False,
        ),
    ] = None,
)

Runs the European options pricing and performance benchmark.

Source code in src/optpricing/cli/commands/demo.py

@app.command(name="european")
def demo_european(
    model: Annotated[
        str | None,
        typer.Option(
            "--model",
            "-m",
            help="Run for a specific model (e.g., 'BSM').",
            case_sensitive=False,
        ),
    ] = None,
    technique: Annotated[
        str | None,
        typer.Option(
            "--technique",
            "-t",
            help="Run for a specific technique (e.g., 'MC').",
            case_sensitive=False,
        ),
    ] = None,
):
    """Runs the European options pricing and performance benchmark."""
    args = []
    if model:
        args.extend(["--model", model])
    if technique:
        args.extend(["--technique", technique])
    _run_example("european_options_benchmark.py", args)

demo_rates

demo_rates()

Runs the interest rate models benchmark.

Source code in src/optpricing/cli/commands/demo.py

@app.command(name="rates")
def demo_rates():
    """Runs the interest rate models benchmark."""
    _run_example("rate_models_benchmark.py", [])

price

price(
    ticker: Annotated[
        str, Option("--ticker", "-t", help="Stock ticker.")
    ],
    strike: Annotated[
        float,
        Option("--strike", "-k", help="Strike price."),
    ],
    maturity: Annotated[
        str,
        Option(
            "--maturity", "-T", help="Maturity YYYY-MM-DD."
        ),
    ],
    option_type: Annotated[
        str,
        Option(
            "--type", help="call|put", case_sensitive=False
        ),
    ] = "call",
    style: Annotated[
        str,
        Option(
            "--style",
            help="european|american",
            case_sensitive=False,
        ),
    ] = "european",
    model: Annotated[
        str, Option("--model", "-m", help="Model key.")
    ] = "BSM",
    technique: Annotated[
        str | None,
        Option(
            "--technique",
            "-x",
            help="Force technique key (e.g., mc, crr).",
            case_sensitive=False,
        ),
    ] = None,
    param: Annotated[
        list[str] | None,
        Option(
            "--param",
            help="Repeat: key=value (e.g. sigma=0.2)",
            show_default=False,
        ),
    ] = None,
)

Prices a single option using live market data and specified parameters.

Source code in src/optpricing/cli/commands/price.py

def price(
    ticker: Annotated[
        str,
        typer.Option(
            "--ticker",
            "-t",
            help="Stock ticker.",
        ),
    ],
    strike: Annotated[
        float,
        typer.Option(
            "--strike",
            "-k",
            help="Strike price.",
        ),
    ],
    maturity: Annotated[
        str,
        typer.Option(
            "--maturity",
            "-T",
            help="Maturity YYYY-MM-DD.",
        ),
    ],
    option_type: Annotated[
        str,
        typer.Option(
            "--type",
            help="call|put",
            case_sensitive=False,
        ),
    ] = "call",
    style: Annotated[
        str,
        typer.Option(
            "--style",
            help="european|american",
            case_sensitive=False,
        ),
    ] = "european",
    model: Annotated[
        str,
        typer.Option(
            "--model",
            "-m",
            help="Model key.",
        ),
    ] = "BSM",
    technique: Annotated[
        str | None,
        typer.Option(
            "--technique",
            "-x",
            help="Force technique key (e.g., mc, crr).",
            case_sensitive=False,
        ),
    ] = None,
    param: Annotated[
        list[str] | None,
        typer.Option(
            "--param",
            help="Repeat: key=value (e.g. sigma=0.2)",
            show_default=False,
        ),
    ] = None,
):
    """
    Prices a single option using live market data and specified parameters.
    """

    def _parse_params(param_list: list[str] | None) -> dict[str, float]:
        params: dict[str, float] = {}
        if not param_list:
            return params
        for raw in param_list:
            if "=" not in raw:
                _err(f"Invalid --param '{raw}'. Expected format key=value.")
            key, value = (tok.strip() for tok in raw.split("=", 1))
            try:
                params[key] = float(value)
            except ValueError:
                _err(f"Could not parse float from '{value}' (param '{key}').")
        return params

    def _select_technique() -> Any:
        technique_map = {
            "MC": MonteCarloTechnique,
            "AMERICAN_MC": AmericanMonteCarloTechnique,
            "CRR": CRRTechnique,
            "LR": LeisenReimerTechnique,
        }
        if technique:
            key = technique.upper()
            if key not in technique_map:
                _err(
                    f"Technique '{technique}' not recognised. "
                    f"Choices: {', '.join(technique_map)}"
                )
            try:
                return technique_map[key](is_american=is_american)
            except TypeError:
                return technique_map[key]()

        if is_american:
            if isinstance(model_instance, BSMModel):
                typer.echo("American style detected. Using Leisen-Reimer lattice.")
                return LeisenReimerTechnique(is_american=True)
            typer.echo("American style detected. Falling back to LSMC (American MC).")
            return AmericanMonteCarloTechnique()

        fastest = select_fastest_technique(model_instance)
        _tech = fastest.__class__.__name__
        typer.echo(f"European style. Auto-selected fastest technique: {_tech}.")
        return fastest

    # Parse & Validate Inputs
    typer.echo(
        f"Pricing a {style} {ticker} {option_type.upper()} "
        f"(K={strike}) exp {maturity} using {model}..."
    )
    model_params = _parse_params(param)
    is_american = style.lower() == "american"

    # Live-Data Fetch
    typer.echo("Fetching live option chain...")
    live_chain = get_live_option_chain(ticker)
    if live_chain is None or live_chain.empty:
        _err(f"No live option chain found for {ticker}.")

    q_div = get_live_dividend_yield(ticker)
    spot = live_chain["spot_price"].iloc[0]
    calls = live_chain[live_chain["optionType"] == "call"]
    puts = live_chain[live_chain["optionType"] == "put"]
    r_rate, _ = fit_rate_and_dividend(calls, puts, spot, q_fixed=q_div)
    typer.echo(
        f"Live Data: Spot {spot:.2f} | Dividend {q_div:.4%} | Implied r {r_rate:.4%}"
    )

    # Build Atoms & Model
    stock = Stock(spot=spot, dividend=q_div)
    rate = Rate(rate=r_rate)
    maturity_years = (
        pd.to_datetime(maturity) - pd.Timestamp.utcnow().tz_localize(None)
    ).days / 365.25
    if maturity_years <= 0:
        _err("Maturity date must be in the future.")

    option = Option(
        strike=strike,
        maturity=maturity_years,
        option_type=OptionType[option_type.upper()],
    )

    try:
        model_cls = ALL_MODEL_CONFIGS[model]["model_class"]
    except KeyError:
        _err(f"Model '{model}' not recognised in ALL_MODEL_CONFIGS.")

    # Merge default params with user-provided params
    full_params = (
        model_cls.default_params.copy() if hasattr(model_cls, "default_params") else {}
    )
    full_params.update(model_params)
    model_instance = model_cls(params=full_params)

    # Technique Selection & Pricing
    technique_instance = _select_technique()

    price_result = technique_instance.price(
        option,
        stock,
        model_instance,
        rate,
        **full_params,
    )

    typer.secho("\n── Pricing Results " + "─" * 38, fg=typer.colors.CYAN)
    typer.echo(f"Price: {price_result.price:.4f}")

    # Greeks
    for greek_name in [
        "Delta",
        "Gamma",
        "Vega",
        "Theta",
        "Rho",
    ]:
        greek_func = getattr(
            technique_instance,
            greek_name.lower(),
            None,
        )
        if callable(greek_func):
            try:
                value = greek_func(
                    option,
                    stock,
                    model_instance,
                    rate,
                    **full_params,
                )
                if isinstance(value, int | float):
                    typer.echo(f"{greek_name}: {value:.4f}")
            except NotImplementedError:
                continue

get_implied_rate

get_implied_rate(
    ticker: Annotated[
        str,
        Option(
            "--ticker",
            "-t",
            help="Stock ticker for the option pair.",
        ),
    ],
    strike: Annotated[
        float,
        Option(
            "--strike",
            "-k",
            help="Strike price of the option pair.",
        ),
    ],
    maturity: Annotated[
        str,
        Option(
            "--maturity",
            "-T",
            help="Maturity date in YYYY-MM-DD format.",
        ),
    ],
)

Calculates the implied risk-free rate from a live call-put pair.

Source code in src/optpricing/cli/commands/tools.py

def get_implied_rate(
    ticker: Annotated[
        str, typer.Option("--ticker", "-t", help="Stock ticker for the option pair.")
    ],
    strike: Annotated[
        float, typer.Option("--strike", "-k", help="Strike price of the option pair.")
    ],
    maturity: Annotated[
        str,
        typer.Option("--maturity", "-T", help="Maturity date in YYYY-MM-DD format."),
    ],
):
    """Calculates the implied risk-free rate from a live call-put pair."""
    typer.echo(
        f"Fetching live prices for {ticker} {strike} options expiring {maturity}..."
    )

    live_chain = get_live_option_chain(ticker)
    if live_chain is None or live_chain.empty:
        typer.secho(
            f"Error: Could not fetch live option chain for {ticker}.",
            fg=typer.colors.RED,
        )
        raise typer.Exit(code=1)

    q = get_live_dividend_yield(ticker)

    maturity_dt = pd.to_datetime(maturity).date()
    chain_for_expiry = live_chain[live_chain["expiry"].dt.date == maturity_dt]

    call_option = chain_for_expiry[
        (chain_for_expiry["strike"] == strike)
        & (chain_for_expiry["optionType"] == "call")
    ]
    put_option = chain_for_expiry[
        (chain_for_expiry["strike"] == strike)
        & (chain_for_expiry["optionType"] == "put")
    ]

    if call_option.empty or put_option.empty:
        typer.secho(
            f"Error: Did not find both: call & put for strike {strike} on {maturity}.",
            fg=typer.colors.RED,
        )
        raise typer.Exit(code=1)

    call_price = call_option["marketPrice"].iloc[0]
    put_price = put_option["marketPrice"].iloc[0]

    spot_price = call_option["spot_price"].iloc[0]
    maturity_years = call_option["maturity"].iloc[0]

    pair_msg = (
        f"Found Pair -> Call Price: {call_price:.2f}, Put Price: {put_price:.2f}, "
        f"Spot: {spot_price:.2f}"
    )
    typer.echo(pair_msg)

    implied_rate_model = ImpliedRateModel(params={})
    try:
        implied_r = implied_rate_model.price_closed_form(
            call_price=call_price,
            put_price=put_price,
            spot=spot_price,
            strike=strike,
            t=maturity_years,
            q=q,
        )
        typer.secho(
            f"\nImplied Risk-Free Rate (r): {implied_r:.4%}", fg=typer.colors.GREEN
        )
    except Exception as e:
        typer.secho(f"\nError calculating implied rate: {e}", fg=typer.colors.RED)