Skip to content

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[list[str] | None, Option('--model', '-m', help="Model to backtest ('all' for every calibratable model). Can be used multiple times.")] = None, auto: Annotated[bool, Option('--auto', help='Use the settings tuned for this ticker (`optpricing tune`).')] = False, allow_mc: Annotated[bool, Option('--allow-mc', help='Also backtest simulation-only models (slow).')] = False, n_starts: Annotated[int, Option('--n-starts', help='Starting points per calibration.')] = 2, verbose: Annotated[bool, Option('--verbose', '-v', help='Enable detailed logging.')] = False)

Runs a walk-forward backtest (calibrate on day t, evaluate on day t+1).

Results are saved per model to artifacts/backtest_logs/ (per-period CSV, bucket CSV and a JSON summary) and summarised in a table.

Source code in src/optpricing/cli/commands/backtest.py 
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def backtest(
    ticker: Annotated[
        str, typer.Option("--ticker", "-t", help="The stock ticker to backtest.")
    ],
    model: Annotated[
        list[str] | None,
        typer.Option(
            "--model",
            "-m",
            help="Model to backtest ('all' for every calibratable model). "
            "Can be used multiple times.",
        ),
    ] = None,
    auto: Annotated[
        bool,
        typer.Option(
            "--auto", help="Use the settings tuned for this ticker (`optpricing tune`)."
        ),
    ] = False,
    allow_mc: Annotated[
        bool,
        typer.Option("--allow-mc", help="Also backtest simulation-only models (slow)."),
    ] = False,
    n_starts: Annotated[
        int, typer.Option("--n-starts", help="Starting points per calibration.")
    ] = 2,
    verbose: Annotated[
        bool, typer.Option("--verbose", "-v", help="Enable detailed logging.")
    ] = False,
):
    """
    Runs a walk-forward backtest (calibrate on day t, evaluate on day t+1).

    Results are saved per model to `artifacts/backtest_logs/` (per-period CSV,
    bucket CSV and a JSON summary) and summarised in a table.
    """
    from optpricing.cli.main import setup_logging

    setup_logging(verbose)
    ticker = ticker.upper()
    models = select_models(model or ["all"], allow_mc=allow_mc)
    base_settings = CalibrationSettings(n_starts=n_starts)

    summaries: list[dict[str, Any]] = []
    for model_name in models:
        from optpricing.workflows.configs import get_model_config

        config = get_model_config(model_name)
        settings = base_settings
        if auto:
            tuned = load_tuned_settings(ticker, model_name)
            if tuned is not None:
                settings = tuned
        typer.echo(f"Backtesting {model_name} ({settings.name} settings)...")
        workflow = BacktestWorkflow(
            ticker, config, settings=settings, allow_mc=allow_mc
        )
        workflow.run()
        saved = workflow.save_results()
        summary = workflow.summary()
        summaries.append(summary)
        if saved:
            typer.echo(f"  - Saved: {saved[0]}")
        elif summary["n_periods"] == 0:
            typer.secho(
                "  - Not enough snapshots for a backtest (need at least two dates).",
                fg=typer.colors.YELLOW,
            )

    summaries.sort(
        key=lambda s: (
            s.get("mean_oos_iv_rmse")
            if s.get("mean_oos_iv_rmse") is not None
            else float("inf")
        )
    )
    print_backtest_summary(summaries, ticker)

print_backtest_summary 

print_backtest_summary(summaries: list[dict[str, Any]], ticker: str) -> None

Rich table of per-model backtest summaries (best out-of-sample first).

Source code in src/optpricing/cli/commands/backtest.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def print_backtest_summary(summaries: list[dict[str, Any]], ticker: str) -> None:
    """Rich table of per-model backtest summaries (best out-of-sample first)."""
    console = Console()
    table = Table(title=f"Walk-forward backtest - {ticker}")
    for col, justify in (
        ("#", "right"),
        ("Model", "left"),
        ("Periods", "right"),
        ("OOS IV RMSE", "right"),
        ("In-sample IV RMSE", "right"),
        ("OOS price RMSE", "right"),
        ("BSM-ATM IV RMSE", "right"),
        ("Beats BSM-ATM", "right"),
        ("Time/period (s)", "right"),
    ):
        table.add_column(col, justify=justify)
    for i, s in enumerate(summaries, start=1):
        table.add_row(
            str(i),
            s["model"],
            f"{s['n_periods'] - s['n_failed']}/{s['n_periods']}",
            _fmt(s.get("mean_oos_iv_rmse")),
            _fmt(s.get("mean_in_sample_iv_rmse")),
            _fmt(s.get("mean_oos_price_rmse")),
            _fmt(s.get("mean_bsm_atm_iv_rmse")),
            _fmt(s.get("share_beating_bsm_atm"), 2),
            _fmt(s.get("mean_runtime_s"), 1),
        )
    console.print(table)

calibrate 

calibrate(ticker: Annotated[str, Option('--ticker', '-t', help='The stock ticker to calibrate against.')], model: Annotated[list[str] | None, Option('--model', '-m', help="Model to calibrate ('all' for every calibratable model). Can be used multiple times.")] = None, 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, auto: Annotated[bool, Option('--auto', help='Use the settings tuned for this ticker (`optpricing tune`).')] = False, allow_mc: Annotated[bool, Option('--allow-mc', help='Also calibrate simulation-only models (slow).')] = False, objective: Annotated[str, Option('--objective', help="Residual space: 'iv' or 'price'.")] = 'iv', n_starts: Annotated[int, Option('--n-starts', help='Starting points per model.')] = 4, verbose: Annotated[bool, Option('--verbose', '-v', help='Enable detailed logging.')] = False)

Calibrates one or more models to a market snapshot and ranks them.

With --model all every model that can be priced on a chain is fitted and a leaderboard (sorted by evaluation IV RMSE) is printed and saved to artifacts/calibration_logs/; parameters go to artifacts/calibrated_params/.

Source code in src/optpricing/cli/commands/calibrate.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
def calibrate(
    ticker: Annotated[
        str,
        typer.Option("--ticker", "-t", help="The stock ticker to calibrate against."),
    ],
    model: Annotated[
        list[str] | None,
        typer.Option(
            "--model",
            "-m",
            help="Model to calibrate ('all' for every calibratable model). "
            "Can be used multiple times.",
        ),
    ] = None,
    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,
    auto: Annotated[
        bool,
        typer.Option(
            "--auto",
            help="Use the settings tuned for this ticker (`optpricing tune`).",
        ),
    ] = False,
    allow_mc: Annotated[
        bool,
        typer.Option(
            "--allow-mc", help="Also calibrate simulation-only models (slow)."
        ),
    ] = False,
    objective: Annotated[
        str,
        typer.Option("--objective", help="Residual space: 'iv' or 'price'."),
    ] = "iv",
    n_starts: Annotated[
        int, typer.Option("--n-starts", help="Starting points per model.")
    ] = 4,
    verbose: Annotated[
        bool, typer.Option("--verbose", "-v", help="Enable detailed logging.")
    ] = False,
):
    """
    Calibrates one or more models to a market snapshot and ranks them.

    With `--model all` every model that can be priced on a chain is fitted and a
    leaderboard (sorted by evaluation IV RMSE) is printed and saved to
    `artifacts/calibration_logs/`; parameters go to `artifacts/calibrated_params/`.
    """
    from optpricing.cli.main import setup_logging

    setup_logging(verbose)
    ticker = ticker.upper()
    models = select_models(model or ["all"], allow_mc=allow_mc)

    if date is None:
        typer.echo(f"No date specified for {ticker}. Finding latest 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 = parse_fixed_params(fix_param)
    try:
        base_settings = CalibrationSettings(objective=objective, n_starts=n_starts)
    except ValueError as e:
        typer.secho(f"Error: {e}", fg=typer.colors.RED)
        raise typer.Exit(code=1)

    CALIBRATED_PARAMS_DIR.mkdir(parents=True, exist_ok=True)
    rows: list[dict[str, Any]] = []
    for model_name in models:
        config = get_model_config(model_name)
        config["ticker"] = ticker
        config["frozen"] = {**config.get("frozen", {}), **frozen_params}
        settings = base_settings
        if auto:
            tuned = load_tuned_settings(ticker, model_name)
            if tuned is not None:
                settings = (
                    tuned.replace(objective=objective)
                    if objective != tuned.objective
                    else tuned
                )
            else:
                typer.secho(
                    f"No tuned settings for {ticker}/{model_name}; using defaults.",
                    fg=typer.colors.YELLOW,
                )

        typer.echo(f"Calibrating {model_name} on {date} ({settings.name} settings)...")
        workflow = DailyWorkflow(
            market_data, config, settings=settings, allow_mc=allow_mc
        )
        workflow.run()
        result = dict(workflow.results)
        result["Settings"] = settings.name
        result.pop("Buckets", None)
        rows.append(result)

        if result["Status"] == "Success":
            params_to_save = {
                "model": model_name,
                "ticker": ticker,
                "date": date,
                "params": {k: float(v) for k, v in result["Calibrated Params"].items()},
                "rate": result["Implied Rate"],
                "dividend_yield": result["Known Dividend"],
                "calibration_iv_rmse": result["Calibration IV RMSE"],
                "evaluation_iv_rmse": result["IV RMSE"],
                "evaluation_price_rmse": result["RMSE"],
                "settings": settings.to_dict(),
            }
            save_path = CALIBRATED_PARAMS_DIR / f"{ticker}_{model_name}_{date}.json"
            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"  - {model_name} FAILED: {result.get('Error', 'Unknown error')}",
                fg=typer.colors.RED,
            )

    rows.sort(
        key=lambda r: (
            r["Status"] != "Success",
            r.get("IV RMSE")
            if pd.notna(r.get("IV RMSE", float("nan")))
            else float("inf"),
        )
    )
    print_leaderboard(rows, f"Calibration leaderboard - {ticker} {date}")
    if any(r["Status"] == "Success" for r in rows):
        CALIBRATION_LOGS_DIR.mkdir(parents=True, exist_ok=True)
        out = CALIBRATION_LOGS_DIR / f"{ticker}_leaderboard_{date}.csv"
        frame = pd.DataFrame(rows)
        if "Calibrated Params" in frame.columns:
            frame["Calibrated Params"] = frame["Calibrated Params"].map(
                lambda p: json.dumps(p) if isinstance(p, dict) else p
            )
        frame.to_csv(out, index=False)
        typer.echo(f"Leaderboard saved to: {out}")
    if not any(r["Status"] == "Success" for r in rows):
        raise typer.Exit(code=1)

parse_fixed_params 

parse_fixed_params(fix_param: list[str] | None) -> dict[str, float]

Parses key=value strings into a dict, exiting with a message on error.

Source code in src/optpricing/cli/commands/calibrate.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def parse_fixed_params(fix_param: list[str] | None) -> dict[str, float]:
    """Parses ``key=value`` strings into a dict, exiting with a message on error."""
    frozen: dict[str, float] = {}
    for p in fix_param or []:
        try:
            key, value = p.split("=", 1)
            frozen[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)
    return frozen

print_leaderboard 

print_leaderboard(rows: list[dict[str, Any]], title: str) -> None

Rich table of per-model calibration results.

Source code in src/optpricing/cli/commands/calibrate.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def print_leaderboard(rows: list[dict[str, Any]], title: str) -> None:
    """Rich table of per-model calibration results."""
    console = Console()
    table = Table(title=title)
    for col, justify in (
        ("#", "right"),
        ("Model", "left"),
        ("Status", "left"),
        ("Calib IV RMSE", "right"),
        ("Eval IV RMSE", "right"),
        ("Price RMSE", "right"),
        ("Options", "right"),
        ("Time (s)", "right"),
        ("Settings", "left"),
    ):
        table.add_column(col, justify=justify)
    for i, r in enumerate(rows, start=1):
        table.add_row(
            str(i),
            str(r["Model"]),
            "OK" if r.get("Status") == "Success" else "FAIL",
            _fmt(r.get("Calibration IV RMSE")),
            _fmt(r.get("IV RMSE")),
            _fmt(r.get("RMSE")),
            f"{r.get('N Calibration Options', 0) or 0}/{r.get('N Options', 0) or 0}",
            _fmt(r.get("Runtime (s)"), 1),
            str(r.get("Settings", "")),
        )
    console.print(table)

select_models 

select_models(model: list[str], *, allow_mc: bool) -> list[str]

Expands all and aliases; exits on unknown names.

Source code in src/optpricing/cli/commands/calibrate.py 
45
46
47
48
49
50
51
52
53
54
55
56
def select_models(model: list[str], *, allow_mc: bool) -> list[str]:
    """Expands ``all`` and aliases; exits on unknown names."""
    if any(m.lower() == "all" for m in model):
        return calibratable_model_names(allow_mc=allow_mc)
    names: list[str] = []
    for m in model:
        try:
            names.append(resolve_model_name(m))
        except KeyError as e:
            typer.secho(str(e.args[0]), fg=typer.colors.RED)
            raise typer.Exit(code=1)
    return names

dashboard 

dashboard()

Launches the Streamlit dashboard application.

Source code in src/optpricing/cli/commands/dashboard.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
def dashboard():
    """
    Launches the Streamlit dashboard application.
    """
    try:
        home = resources.files("optpricing.dashboard").joinpath("Home.py")
        with resources.as_file(home) 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[ui]'`",
            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', rates: Annotated[bool, Option('--rates', help='Also download the ^IRX (13-week T-bill) history used as the risk-free rate of older snapshots.')] = False)

Downloads historical prices, dividends and log returns (and optionally the risk-free rate history) for specified tickers or all defaults.

Source code in src/optpricing/cli/commands/data.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
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",
    rates: Annotated[
        bool,
        typer.Option(
            "--rates",
            help="Also download the ^IRX (13-week T-bill) history used as the "
            "risk-free rate of older snapshots.",
        ),
    ] = False,
):
    """
    Downloads historical prices, dividends and log returns (and optionally the
    risk-free rate history) for specified tickers or all defaults.
    """
    if rates:
        typer.echo(f"Downloading {period} risk-free (^IRX) history...")
        save_risk_free_history(period=period)
        if not tickers and not all_default:
            typer.secho("Download complete.", fg=typer.colors.GREEN)
            return
    tickers_to_process = _resolve_tickers(
        tickers, all_default, f"Downloading {period} historical data"
    )
    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 
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
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.
    """
    tickers_to_fetch = _resolve_tickers(
        tickers, all_default, "Fetching dividend yields"
    )

    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, compact: Annotated[bool, Option('--compact', help='Keep only liquid options and compress (what the scheduled GitHub Action commits; ~10x smaller files).')] = False, only_fresh: Annotated[bool, Option('--only-fresh', help='Skip tickers whose chain shows no trade today (holidays).')] = False)

Fetches and saves a live market data snapshot (schema v2: spot, rate and dividend yield included) for specified tickers.

Source code in src/optpricing/cli/commands/data.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
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,
    compact: Annotated[
        bool,
        typer.Option(
            "--compact",
            help="Keep only liquid options and compress (what the scheduled "
            "GitHub Action commits; ~10x smaller files).",
        ),
    ] = False,
    only_fresh: Annotated[
        bool,
        typer.Option(
            "--only-fresh",
            help="Skip tickers whose chain shows no trade today (holidays).",
        ),
    ] = False,
):
    """
    Fetches and saves a live market data snapshot (schema v2: spot, rate and
    dividend yield included) for specified tickers.
    """
    tickers_to_process = _resolve_tickers(
        tickers, all_default, "Saving live market snapshots"
    )
    written = save_market_snapshot(
        tickers_to_process, compact=compact, only_fresh=only_fresh
    )
    for path in written:
        typer.echo(f"  - {path}")
    if not written:
        typer.secho("No snapshot was written.", fg=typer.colors.YELLOW)
        raise typer.Exit(code=1)
    typer.secho(f"Snapshot complete ({len(written)} file(s)).", fg=typer.colors.GREEN)

stream_quotes 

stream_quotes(tickers: Annotated[list[str] | None, Option('--ticker', '-t', help='Underlying to stream (repeatable).')] = None, symbols: Annotated[list[str] | None, Option('--symbol', '-s', help="Any extra Yahoo symbol, e.g. an option contract 'SPY250919C00600000' or an index '^SPX' (repeatable).")] = None, duration: Annotated[float, Option('--duration', '-d', help='Recording length in seconds.')] = 60.0, contracts: Annotated[int, Option('--contracts', '-c', help='Also stream the N most active option contracts of each ticker.')] = 0, output: Annotated[Path | None, Option('--output', '-o', help='Parquet file (default: data/live_quotes/).')] = None)

Records live quotes from Yahoo Finance's free streamer into a parquet file.

Press Ctrl-C to stop early; whatever was received is saved.

Source code in src/optpricing/cli/commands/data.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
def stream_quotes(
    tickers: Annotated[
        list[str] | None,
        typer.Option(
            "--ticker",
            "-t",
            help="Underlying to stream (repeatable).",
        ),
    ] = None,
    symbols: Annotated[
        list[str] | None,
        typer.Option(
            "--symbol",
            "-s",
            help="Any extra Yahoo symbol, e.g. an option contract "
            "'SPY250919C00600000' or an index '^SPX' (repeatable).",
        ),
    ] = None,
    duration: Annotated[
        float,
        typer.Option("--duration", "-d", help="Recording length in seconds."),
    ] = 60.0,
    contracts: Annotated[
        int,
        typer.Option(
            "--contracts",
            "-c",
            help="Also stream the N most active option contracts of each ticker.",
        ),
    ] = 0,
    output: Annotated[
        Path | None,
        typer.Option(
            "--output", "-o", help="Parquet file (default: data/live_quotes/)."
        ),
    ] = None,
):
    """
    Records live quotes from Yahoo Finance's free streamer into a parquet file.

    Press Ctrl-C to stop early; whatever was received is saved.
    """
    requested = [t.upper() for t in (tickers or [])] + [
        s.upper() for s in (symbols or [])
    ]
    if not requested:
        typer.secho(
            "Error: provide at least one --ticker or --symbol.", fg=typer.colors.RED
        )
        raise typer.Exit(code=1)
    if contracts > 0:
        for t in tickers or []:
            found = option_contract_symbols(t, contracts)
            typer.echo(f"  {t}: streaming {len(found)} option contract(s)")
            requested.extend(found)
    seen: set[str] = set()
    all_symbols = [s for s in requested if not (s in seen or seen.add(s))]
    typer.echo(f"Streaming {len(all_symbols)} symbol(s) for {duration:.0f}s...")
    path, n = record_live_quotes(all_symbols, duration, output)
    if path is None:
        typer.secho("No quotes received (market closed?).", fg=typer.colors.YELLOW)
        raise typer.Exit(code=1)
    typer.secho(f"Recorded {n} quotes to {path}", fg=typer.colors.GREEN)

demo_american 

demo_american()

Runs the American options pricing benchmark.

Source code in src/optpricing/cli/commands/demo.py 
77
78
79
80
@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 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
@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 
83
84
85
86
@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 
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
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}.")

    # Spot, T-bill rate and parity-implied dividend yield travel with the chain.
    stock, rate = snapshot_market_params(live_chain)
    spot, q_div, r_rate = stock.spot, stock.dividend, rate.get_rate()
    typer.echo(
        f"Live Data: Spot {spot:.2f} | Implied dividend {q_div:.4%} | r {r_rate:.4%}"
    )

    maturity_years = (
        pd.to_datetime(maturity) - pd.Timestamp.now(tz="UTC").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 = get_model_config(model)["model_class"]
    except KeyError as e:
        _err(str(e.args[0]))

    # 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()

    try:
        price_result = technique_instance.price(
            option,
            stock,
            model_instance,
            rate,
            **full_params,
        )
    except TypeError as e:
        _err(str(e))

    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 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
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)

print_tuning_leaderboard 

print_tuning_leaderboard(result: TuningResult) -> None

Rich table of the tuning leaderboard (best model first).

Source code in src/optpricing/cli/commands/tune.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def print_tuning_leaderboard(result: TuningResult) -> None:
    """Rich table of the tuning leaderboard (best model first)."""
    console = Console()
    table = Table(
        title=f"Tuning leaderboard - {result.ticker} "
        f"({len(result.dates)} snapshots, {result.dates[0]} to {result.dates[-1]})"
    )
    for col, justify in (
        ("#", "right"),
        ("Model", "left"),
        ("Best settings", "left"),
        ("OOS IV RMSE", "right"),
        ("In-sample IV RMSE", "right"),
        ("OOS price RMSE", "right"),
        ("BSM-ATM IV RMSE", "right"),
        ("Periods", "right"),
        ("Time (s)", "right"),
    ):
        table.add_column(col, justify=justify)
    for row in result.leaderboard:
        table.add_row(
            str(row.get("rank", "")),
            row["model"],
            row["settings"],
            f"{row['oos_iv_rmse']:.4f}",
            f"{row['in_sample_iv_rmse']:.4f}"
            if pd.notna(row.get("in_sample_iv_rmse"))
            else "-",
            f"{row['oos_price_rmse']:.3f}"
            if pd.notna(row.get("oos_price_rmse"))
            else "-",
            f"{row['bsm_atm_iv_rmse']:.4f}"
            if pd.notna(row.get("bsm_atm_iv_rmse"))
            else "-",
            f"{row['n_periods'] - row['n_failed']}/{row['n_periods']}",
            f"{row['runtime_s']:.1f}",
        )
    console.print(table)

tune 

tune(ticker: Annotated[str, Option('--ticker', '-t', help='The stock ticker to tune for.')], model: Annotated[list[str] | None, Option('--model', '-m', help="Model to include ('all' by default). Can be used multiple times.")] = None, candidates: Annotated[str, Option('--candidates', help="'quick' (3 settings), 'full' (all) or a comma-separated list of: " + join(CANDIDATE_SETTINGS))] = 'quick', max_pairs: Annotated[int | None, Option('--max-pairs', help='Use only the most recent N day pairs.')] = None, allow_mc: Annotated[bool, Option('--allow-mc', help='Include simulation-only models (slow).')] = False, verbose: Annotated[bool, Option('--verbose', '-v', help='Enable detailed logging.')] = False)

Tunes calibration settings per model by walk-forward cross-validation and ranks the models for a ticker.

The result is cached in artifacts/tuning/{TICKER}.json and used by optpricing calibrate --auto and optpricing backtest --auto.

Source code in src/optpricing/cli/commands/tune.py 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def tune(
    ticker: Annotated[
        str, typer.Option("--ticker", "-t", help="The stock ticker to tune for.")
    ],
    model: Annotated[
        list[str] | None,
        typer.Option(
            "--model",
            "-m",
            help="Model to include ('all' by default). Can be used multiple times.",
        ),
    ] = None,
    candidates: Annotated[
        str,
        typer.Option(
            "--candidates",
            help="'quick' (3 settings), 'full' (all) or a comma-separated list of: "
            + ", ".join(CANDIDATE_SETTINGS),
        ),
    ] = "quick",
    max_pairs: Annotated[
        int | None,
        typer.Option("--max-pairs", help="Use only the most recent N day pairs."),
    ] = None,
    allow_mc: Annotated[
        bool,
        typer.Option("--allow-mc", help="Include simulation-only models (slow)."),
    ] = False,
    verbose: Annotated[
        bool, typer.Option("--verbose", "-v", help="Enable detailed logging.")
    ] = False,
):
    """
    Tunes calibration settings per model by walk-forward cross-validation and
    ranks the models for a ticker.

    The result is cached in `artifacts/tuning/{TICKER}.json` and used by
    `optpricing calibrate --auto` and `optpricing backtest --auto`.
    """
    from optpricing.cli.main import setup_logging

    setup_logging(verbose)
    models = select_models(model or ["all"], allow_mc=allow_mc)
    candidate_arg: str | list[str]
    if candidates in ("quick", "full"):
        candidate_arg = candidates
    else:
        candidate_arg = [c.strip() for c in candidates.split(",") if c.strip()]

    try:
        result = tune_ticker(
            ticker,
            models=models,
            candidates=candidate_arg,
            max_pairs=max_pairs,
            allow_mc=allow_mc,
            progress=lambda msg: typer.echo(f"  tuning {msg}..."),
        )
    except ValueError as e:
        typer.secho(f"Error: {e}", fg=typer.colors.RED)
        raise typer.Exit(code=1)

    if not result.leaderboard:
        typer.secho("No model could be tuned.", fg=typer.colors.RED)
        raise typer.Exit(code=1)
    print_tuning_leaderboard(result)
    typer.secho(
        f"Best model for {result.ticker}: {result.best_model}. "
        f"Settings cached in {result.save()}",
        fg=typer.colors.GREEN,
    )