gavindiaz/ferro-ta
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: update ferro-ta version to 1.1.3

Browse Source 
- Bumped version numbers across Cargo.toml, Cargo.lock, pyproject.toml, and conda/meta.yaml to 1.1.3.
- Added new features including American option pricing, digital options, extended Greeks, and historical volatility estimators.
- Enhanced documentation and tests for new functionalities.
- Updated CHANGELOG.md to reflect changes for version 1.1.3.
This commit is contained in:
Pratik Bhadane
2026-04-02 16:30:45 +05:30
parent 125eb32d9f
commit 2080e4673d
38 changed files with 5619 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/ferro_ta/analysis/derivatives_payoff.py
+155 -5
View File
@@ -14,6 +14,7 @@ from numpy.typing import ArrayLike, NDArray
from ferro_ta._ferro_ta import aggregate_greeks_legs as _rust_aggregate_greeks_legs
from ferro_ta._ferro_ta import strategy_payoff_dense as _rust_strategy_payoff_dense
from ferro_ta._ferro_ta import strategy_payoff_legs as _rust_strategy_payoff_legs
from ferro_ta._ferro_ta import strategy_value_dense as _rust_strategy_value_dense
from ferro_ta.analysis.options import OptionGreeks
from ferro_ta.analysis.options_strategy import DerivativesStrategy, StrategyLeg
from ferro_ta.core.exceptions import (
@@ -26,7 +27,9 @@ __all__ = [
"PayoffLeg",
"option_leg_payoff",
"futures_leg_payoff",
"stock_leg_payoff",
"strategy_payoff",
"strategy_value",
"aggregate_greeks",
]
@@ -47,8 +50,10 @@ class PayoffLeg:
multiplier: float = 1.0
def __post_init__(self) -> None:
if self.instrument not in {"option", "future"}:
raise FerroTAValueError("instrument must be 'option' or 'future'.")
if self.instrument not in {"option", "future", "stock"}:
raise FerroTAValueError(
"instrument must be 'option', 'future', or 'stock'."
)
if self.side not in {"long", "short"}:
raise FerroTAValueError("side must be 'long' or 'short'.")
if self.instrument == "option":
@@ -58,8 +63,8 @@ class PayoffLeg:
)
if self.strike is None:
raise FerroTAValueError("option legs require strike.")
if self.instrument == "future" and self.entry_price is None:
raise FerroTAValueError("future legs require entry_price.")
if self.instrument in {"future", "stock"} and self.entry_price is None:
raise FerroTAValueError(f"{self.instrument} legs require entry_price.")
def _side_sign(side: str) -> float:
@@ -131,6 +136,60 @@ def futures_leg_payoff(
)
def stock_leg_payoff(
spot_grid: ArrayLike,
*,
entry_price: float,
side: str = "long",
quantity: float = 1.0,
multiplier: float = 1.0,
) -> NDArray[np.float64]:
"""P/L profile for a single stock (equity) leg over a spot grid.
Payoff is linear::
P/L = sign(side) × quantity × multiplier × (spot − entry_price)
Mathematically equivalent to a futures leg — no optionality. Use this
leg type when modelling strategies that hold the underlying equity:
Covered Call, Protective Put, Collar, Covered Strangle, etc.
Parameters
----------
spot_grid:
1-D array of spot prices at which to evaluate the P/L.
entry_price:
Purchase (or short-sale) price of the stock.
side:
``"long"`` (default) or ``"short"``.
quantity:
Number of shares / contracts (default 1).
multiplier:
Contract multiplier (default 1.0).
Returns
-------
NDArray[float64]
P/L at each grid point, same shape as *spot_grid*.
"""
grid = _coerce_spot_grid(spot_grid)
_side_sign(side)
return np.asarray(
_rust_strategy_payoff_dense(
grid,
np.array([2], dtype=np.int64), # stock
np.array([1 if side == "long" else -1], dtype=np.int64),
np.array([-1], dtype=np.int64),
np.array([0.0], dtype=np.float64),
np.array([0.0], dtype=np.float64),
np.array([float(entry_price)], dtype=np.float64),
np.array([float(quantity)], dtype=np.float64),
np.array([float(multiplier)], dtype=np.float64),
),
dtype=np.float64,
)
def _mapping_to_leg(mapping: Mapping[str, Any]) -> PayoffLeg:
return PayoffLeg(**mapping)
@@ -141,7 +200,9 @@ def _strategy_leg_to_payoff_leg(leg: StrategyLeg) -> PayoffLeg:
side=leg.side,
quantity=float(leg.quantity),
option_type=leg.option_type,
strike=leg.strike_selector.explicit_strike,
strike=leg.strike_selector.explicit_strike
if leg.strike_selector is not None
else None,
)
@@ -205,3 +266,92 @@ def aggregate_greeks(
float(theta),
float(rho),
)
def strategy_value(
spot_grid: ArrayLike,
*,
legs: Sequence[PayoffLeg | Mapping[str, Any]],
time_to_expiry: float,
volatility: float,
rate: float = 0.0,
carry: float = 0.0,
) -> NDArray[np.float64]:
"""Current BSM mid-price value of a multi-leg strategy over a spot grid.
Unlike :func:`strategy_payoff` (which computes intrinsic value at expiry),
this uses live BSM pricing for option legs so the result reflects the
pre-expiry value including time value.
Parameters
----------
spot_grid:
Array of spot prices to evaluate.
legs:
Sequence of :class:`PayoffLeg` (or dicts). Option legs must have
``strike`` and ``premium`` set; future/stock legs must have
``entry_price`` set.
time_to_expiry:
Shared time-to-expiry (years) applied to all option legs.
volatility:
Shared implied vol applied to all option legs.
rate:
Risk-free rate applied to all legs.
carry:
Carry / dividend yield applied to all option legs.
"""
grid = _coerce_spot_grid(spot_grid)
normalized: tuple[PayoffLeg, ...] = tuple(
leg if isinstance(leg, PayoffLeg) else _mapping_to_leg(leg) for leg in legs
)
if len(normalized) == 0:
return np.zeros_like(grid)
n_legs = len(normalized)
instruments = np.empty(n_legs, dtype=np.int64)
sides = np.empty(n_legs, dtype=np.int64)
option_types = np.empty(n_legs, dtype=np.int64)
strikes = np.zeros(n_legs, dtype=np.float64)
premiums = np.zeros(n_legs, dtype=np.float64)
entry_prices = np.zeros(n_legs, dtype=np.float64)
quantities = np.ones(n_legs, dtype=np.float64)
multipliers = np.ones(n_legs, dtype=np.float64)
ttes = np.full(n_legs, time_to_expiry, dtype=np.float64)
vols = np.full(n_legs, volatility, dtype=np.float64)
rates = np.full(n_legs, rate, dtype=np.float64)
carries = np.full(n_legs, carry, dtype=np.float64)
_inst_map = {"option": 0, "future": 1, "stock": 2}
for i, leg in enumerate(normalized):
instruments[i] = _inst_map[leg.instrument]
sides[i] = 1 if leg.side == "long" else -1
option_types[i] = 1 if leg.option_type == "call" else -1
if leg.strike is not None:
strikes[i] = float(leg.strike)
premiums[i] = float(leg.premium)
if leg.entry_price is not None:
entry_prices[i] = float(leg.entry_price)
quantities[i] = float(leg.quantity)
multipliers[i] = float(leg.multiplier)
try:
return np.asarray(
_rust_strategy_value_dense(
grid,
instruments,
sides,
option_types,
strikes,
premiums,
entry_prices,
quantities,
multipliers,
ttes,
vols,
rates,
carries,
),
dtype=np.float64,
)
except ValueError as err:
_normalize_rust_error(err)
python/ferro_ta/analysis/options.py
+963
View File
@@ -26,6 +26,15 @@ from ferro_ta._ferro_ta import (
from ferro_ta._ferro_ta import (
bsm_price_batch as _rust_bsm_price_batch,
)
from ferro_ta._ferro_ta import (
expected_move as _rust_expected_move,
)
from ferro_ta._ferro_ta import (
extended_greeks as _rust_extended_greeks,
)
from ferro_ta._ferro_ta import (
extended_greeks_batch as _rust_extended_greeks_batch,
)
from ferro_ta._ferro_ta import (
implied_volatility as _rust_implied_volatility,
)
@@ -50,6 +59,9 @@ from ferro_ta._ferro_ta import (
from ferro_ta._ferro_ta import (
option_greeks_batch as _rust_option_greeks_batch,
)
from ferro_ta._ferro_ta import (
put_call_parity_deviation as _rust_put_call_parity_deviation,
)
from ferro_ta._ferro_ta import (
select_strike_delta as _rust_select_strike_delta,
)
@@ -73,11 +85,14 @@ ScalarOrArray: TypeAlias = float | NDArray[np.float64]
__all__ = [
"OptionGreeks",
"ExtendedGreeks",
"SmileMetrics",
"VolCone",
"black_scholes_price",
"black_76_price",
"option_price",
"greeks",
"extended_greeks",
"implied_volatility",
"smile_metrics",
"term_structure_slope",
@@ -86,9 +101,63 @@ __all__ = [
"iv_rank",
"iv_percentile",
"iv_zscore",
"put_call_parity_deviation",
"expected_move",
"digital_option_price",
"digital_option_greeks",
"american_option_price",
"early_exercise_premium",
"close_to_close_vol",
"parkinson_vol",
"garman_klass_vol",
"rogers_satchell_vol",
"yang_zhang_vol",
"vol_cone",
]
@dataclass(frozen=True)
class ExtendedGreeks:
"""Container for second-order and cross Greeks."""
vanna: ScalarOrArray
volga: ScalarOrArray
charm: ScalarOrArray
speed: ScalarOrArray
color: ScalarOrArray
def to_dict(self) -> dict[str, ScalarOrArray]:
return {
"vanna": self.vanna,
"volga": self.volga,
"charm": self.charm,
"speed": self.speed,
"color": self.color,
}
@dataclass(frozen=True)
class VolCone:
"""Historical realized vol distribution across window lengths."""
windows: NDArray[np.float64]
min: NDArray[np.float64]
p25: NDArray[np.float64]
median: NDArray[np.float64]
p75: NDArray[np.float64]
max: NDArray[np.float64]
def to_dict(self) -> dict[str, NDArray[np.float64]]:
return {
"windows": self.windows,
"min": self.min,
"p25": self.p25,
"median": self.median,
"p75": self.p75,
"max": self.max,
}
@dataclass(frozen=True)
class OptionGreeks:
"""Container for first-order Greeks."""
@@ -630,3 +699,897 @@ def select_strike(
except ValueError as err:
_normalize_rust_error(err)
return None if strike is None else float(strike)
def extended_greeks(
underlying: ArrayLike | float,
strike: ArrayLike | float,
rate: ArrayLike | float,
time_to_expiry: ArrayLike | float,
volatility: ArrayLike | float,
*,
option_type: str = "call",
model: str = "bsm",
carry: ArrayLike | float = 0.0,
) -> ExtendedGreeks:
"""Return vanna, volga, charm, speed, and color (second-order / cross Greeks).
All Greeks are computed via closed-form BSM formulas. Black-76 is not
yet supported and returns NaN for all five values.
Parameters
----------
underlying:
Current underlying (spot) price.
strike:
Option strike price.
rate:
Risk-free rate (annualised, decimal — e.g. ``0.05`` for 5 %).
time_to_expiry:
Time to expiry in years.
volatility:
Implied volatility (annualised, decimal).
option_type:
``"call"`` (default) or ``"put"``.
model:
``"bsm"`` (default). ``"black76"`` returns NaN for all fields.
carry:
Continuous carry / dividend yield (annualised, decimal). Default 0.
Returns
-------
ExtendedGreeks
Named tuple with fields:
- **vanna** — ∂Δ/∂σ: sensitivity of delta to a change in vol.
- **volga** — ∂²V/∂σ² (vomma): sensitivity of vega to a change in vol.
- **charm** — ∂Δ/∂t: daily rate of change in delta (theta of delta).
- **speed** — ∂Γ/∂S: rate of change in gamma with respect to spot.
- **color** — ∂Γ/∂t: daily rate of change in gamma.
Notes
-----
Inputs may be scalars or broadcastable arrays. When arrays are supplied
each field of the returned :class:`ExtendedGreeks` is an ``NDArray``.
Closed-form expressions (BSM, zero-carry)::
vanna = -e^{-qT} · φ(d₁) · d₂ / σ
volga = S · e^{-qT} · φ(d₁) · √T · d₁ · d₂ / σ
charm = -e^{-qT} · φ(d₁) · [2(r-q)T - d₂·σ·√T] / (2T·σ·√T)
speed = -Γ/S · (d₁/(σ√T) + 1)
color = -Γ · [r-q + d₁·σ/(2√T) + (2(r-q)T - d₂·σ√T)·d₁/(2T·σ√T)]
"""
option_type = _validate_option_type(option_type)
model = _validate_model(model)
arrays, scalar_mode = _broadcast_inputs(
underlying=underlying,
strike=strike,
rate=rate,
time_to_expiry=time_to_expiry,
volatility=volatility,
carry=carry,
)
try:
if scalar_mode:
vanna, volga, charm, speed, color = _rust_extended_greeks(
float(arrays["underlying"][0]),
float(arrays["strike"][0]),
float(arrays["rate"][0]),
float(arrays["time_to_expiry"][0]),
float(arrays["volatility"][0]),
option_type,
model,
float(arrays["carry"][0]),
)
return ExtendedGreeks(vanna, volga, charm, speed, color)
vanna, volga, charm, speed, color = _rust_extended_greeks_batch(
arrays["underlying"],
arrays["strike"],
arrays["rate"],
arrays["time_to_expiry"],
arrays["volatility"],
option_type,
model,
arrays["carry"],
)
return ExtendedGreeks(
np.asarray(vanna, dtype=np.float64),
np.asarray(volga, dtype=np.float64),
np.asarray(charm, dtype=np.float64),
np.asarray(speed, dtype=np.float64),
np.asarray(color, dtype=np.float64),
)
except ValueError as err:
_normalize_rust_error(err)
def put_call_parity_deviation(
call_price: float,
put_price: float,
spot: float,
strike: float,
rate: float,
time_to_expiry: float,
*,
carry: float = 0.0,
) -> float:
"""Put-call parity deviation: ``C − P − (S·e^{−q·T} − K·e^{−r·T})``.
At no-arbitrage the deviation is exactly 0. A non-zero result indicates
mispricing, a data error, or a stale quote.
Parameters
----------
call_price:
Market or model price of the call option.
put_price:
Market or model price of the put option.
spot:
Current underlying price.
strike:
Common strike price of the call and put.
rate:
Risk-free rate (annualised, decimal).
time_to_expiry:
Time to expiry in years.
carry:
Continuous dividend yield / carry rate (annualised, decimal).
Returns
-------
float
Signed deviation. Positive → call is overpriced relative to put;
negative → put is overpriced relative to call.
Examples
--------
>>> from ferro_ta.analysis.options import option_price, put_call_parity_deviation
>>> call = option_price(100, 100, 0.05, 1.0, 0.2, option_type="call")
>>> put = option_price(100, 100, 0.05, 1.0, 0.2, option_type="put")
>>> put_call_parity_deviation(call, put, 100, 100, 0.05, 1.0) # ≈ 0.0
"""
try:
return float(
_rust_put_call_parity_deviation(
float(call_price),
float(put_price),
float(spot),
float(strike),
float(rate),
float(time_to_expiry),
float(carry),
)
)
except ValueError as err:
_normalize_rust_error(err)
def expected_move(
spot: float,
iv: float,
days_to_expiry: float,
trading_days_per_year: float = 252.0,
) -> tuple[float, float]:
"""Expected ±1σ move over *days_to_expiry* calendar days.
Uses the log-normal approximation::
upper_move = spot × e^{+σ√(days/trading_days)} − spot
lower_move = spot × e^{−σ√(days/trading_days)} − spot
Parameters
----------
spot:
Current underlying price.
iv:
Implied volatility (annualised, decimal — e.g. ``0.20`` for 20 %).
days_to_expiry:
Number of calendar days until expiry.
trading_days_per_year:
Annualisation factor (default 252).
Returns
-------
tuple[float, float]
``(lower_move, upper_move)`` — signed absolute price changes from
``spot``. ``lower_move < 0``, ``upper_move > 0``.
Notes
-----
Because of log-normal skew, ``|upper_move| > |lower_move|``.
Examples
--------
>>> from ferro_ta.analysis.options import expected_move
>>> lower, upper = expected_move(100.0, 0.20, 30)
>>> round(upper, 2)
7.14
"""
try:
lower, upper = _rust_expected_move(
float(spot), float(iv), float(days_to_expiry), float(trading_days_per_year)
)
return float(lower), float(upper)
except ValueError as err:
_normalize_rust_error(err)
# ---------------------------------------------------------------------------
# Digital options — populated once the Rust bridge is built
# ---------------------------------------------------------------------------
def digital_option_price(
underlying: ArrayLike | float,
strike: ArrayLike | float,
rate: ArrayLike | float,
time_to_expiry: ArrayLike | float,
volatility: ArrayLike | float,
*,
option_type: str = "call",
digital_type: str = "cash_or_nothing",
carry: ArrayLike | float = 0.0,
) -> ScalarOrArray:
"""Price a digital (binary) option under BSM.
Parameters
----------
underlying:
Current underlying (spot) price.
strike:
Option strike price.
rate:
Risk-free rate (annualised, decimal).
time_to_expiry:
Time to expiry in years.
volatility:
Implied volatility (annualised, decimal).
option_type:
``"call"`` (default) or ``"put"``.
digital_type:
``"cash_or_nothing"`` (default) — pays 1 unit of cash if ITM at
expiry; or ``"asset_or_nothing"`` — pays the underlying asset price.
carry:
Continuous carry / dividend yield (annualised, decimal). Default 0.
Returns
-------
float or NDArray[float64]
Option price. Returns a scalar when all inputs are scalars, or an
array when any input is an array.
Notes
-----
Closed-form BSM formulas::
Cash-or-nothing call: e^{−rT} · N(d₂)
Cash-or-nothing put: e^{−rT} · N(−d₂)
Asset-or-nothing call: S · e^{−qT} · N(d₁)
Asset-or-nothing put: S · e^{−qT} · N(−d₁)
Put-call parity for cash-or-nothing: call + put = e^{−rT}.
Put-call parity for asset-or-nothing: call + put = S · e^{−qT}.
Invalid inputs (non-positive spot/strike, negative time or vol) return NaN.
"""
from ferro_ta._ferro_ta import digital_price as _rust_digital_price
from ferro_ta._ferro_ta import digital_price_batch as _rust_digital_price_batch
option_type = _validate_option_type(option_type)
digital_type = digital_type.lower().replace("-", "_")
if digital_type not in {"cash_or_nothing", "asset_or_nothing"}:
raise FerroTAValueError(
"digital_type must be 'cash_or_nothing' or 'asset_or_nothing'."
)
arrays, scalar_mode = _broadcast_inputs(
underlying=underlying,
strike=strike,
rate=rate,
time_to_expiry=time_to_expiry,
volatility=volatility,
carry=carry,
)
try:
if scalar_mode:
return float(
_rust_digital_price(
float(arrays["underlying"][0]),
float(arrays["strike"][0]),
float(arrays["rate"][0]),
float(arrays["time_to_expiry"][0]),
float(arrays["volatility"][0]),
option_type,
digital_type,
float(arrays["carry"][0]),
)
)
out = _rust_digital_price_batch(
arrays["underlying"],
arrays["strike"],
arrays["rate"],
arrays["time_to_expiry"],
arrays["volatility"],
option_type,
digital_type,
arrays["carry"],
)
return np.asarray(out, dtype=np.float64)
except ValueError as err:
_normalize_rust_error(err)
def digital_option_greeks(
underlying: ArrayLike | float,
strike: ArrayLike | float,
rate: ArrayLike | float,
time_to_expiry: ArrayLike | float,
volatility: ArrayLike | float,
*,
option_type: str = "call",
digital_type: str = "cash_or_nothing",
carry: ArrayLike | float = 0.0,
) -> OptionGreeks:
"""Delta, gamma, and vega for a digital option via numerical bumping.
Uses central finite differences (spot bump ε = spot × 10⁻³ for delta/gamma;
vol bump ε = 10⁻³ for vega). Theta and rho are set to NaN.
Parameters
----------
underlying, strike, rate, time_to_expiry, volatility, option_type, carry:
Same as :func:`digital_option_price`.
digital_type:
``"cash_or_nothing"`` (default) or ``"asset_or_nothing"``.
Returns
-------
OptionGreeks
Named tuple; only ``delta``, ``gamma``, ``vega`` are finite.
``theta`` and ``rho`` are NaN.
"""
from ferro_ta._ferro_ta import digital_greeks as _rust_digital_greeks
from ferro_ta._ferro_ta import digital_greeks_batch as _rust_digital_greeks_batch
option_type = _validate_option_type(option_type)
digital_type = digital_type.lower().replace("-", "_")
if digital_type not in {"cash_or_nothing", "asset_or_nothing"}:
raise FerroTAValueError(
"digital_type must be 'cash_or_nothing' or 'asset_or_nothing'."
)
arrays, scalar_mode = _broadcast_inputs(
underlying=underlying,
strike=strike,
rate=rate,
time_to_expiry=time_to_expiry,
volatility=volatility,
carry=carry,
)
try:
if scalar_mode:
delta, gamma, vega = _rust_digital_greeks(
float(arrays["underlying"][0]),
float(arrays["strike"][0]),
float(arrays["rate"][0]),
float(arrays["time_to_expiry"][0]),
float(arrays["volatility"][0]),
option_type,
digital_type,
float(arrays["carry"][0]),
)
return OptionGreeks(delta, gamma, vega, float("nan"), float("nan"))
delta, gamma, vega = _rust_digital_greeks_batch(
arrays["underlying"],
arrays["strike"],
arrays["rate"],
arrays["time_to_expiry"],
arrays["volatility"],
option_type,
digital_type,
arrays["carry"],
)
nan_arr = np.full_like(delta, float("nan"))
return OptionGreeks(
np.asarray(delta, dtype=np.float64),
np.asarray(gamma, dtype=np.float64),
np.asarray(vega, dtype=np.float64),
nan_arr,
nan_arr,
)
except ValueError as err:
_normalize_rust_error(err)
# ---------------------------------------------------------------------------
# American options — populated once the Rust bridge is built
# ---------------------------------------------------------------------------
def american_option_price(
underlying: ArrayLike | float,
strike: ArrayLike | float,
rate: ArrayLike | float,
time_to_expiry: ArrayLike | float,
volatility: ArrayLike | float,
*,
option_type: str = "call",
carry: ArrayLike | float = 0.0,
) -> ScalarOrArray:
"""American option price using the Barone-Adesi-Whaley (1987) approximation.
Accurate to within a few basis points for standard equity/index parameters.
O(1) per evaluation — suitable for batch pricing or calibration.
Parameters
----------
underlying:
Current underlying (spot) price.
strike:
Option strike price.
rate:
Risk-free rate (annualised, decimal).
time_to_expiry:
Time to expiry in years.
volatility:
Implied volatility (annualised, decimal).
option_type:
``"call"`` (default) or ``"put"``.
carry:
Continuous carry / dividend yield (annualised, decimal). Default 0.
For calls with ``carry = 0`` (no dividends) early exercise is never
optimal and the result equals the European BSM price.
Returns
-------
float or NDArray[float64]
American option price ≥ European BSM price.
Notes
-----
The BAW approximation uses a quadratic equation to find the critical
exercise boundary S* via Newton-Raphson iteration, then adds the early
exercise premium on top of the European price.
Reference: Barone-Adesi, G. & Whaley, R.E. (1987). "Efficient Analytic
Approximation of American Option Values." *Journal of Finance*, 42(2),
301–320.
See Also
--------
early_exercise_premium : Difference between American and European prices.
"""
from ferro_ta._ferro_ta import american_price as _rust_american_price
from ferro_ta._ferro_ta import american_price_batch as _rust_american_price_batch
option_type = _validate_option_type(option_type)
arrays, scalar_mode = _broadcast_inputs(
underlying=underlying,
strike=strike,
rate=rate,
time_to_expiry=time_to_expiry,
volatility=volatility,
carry=carry,
)
try:
if scalar_mode:
return float(
_rust_american_price(
float(arrays["underlying"][0]),
float(arrays["strike"][0]),
float(arrays["rate"][0]),
float(arrays["time_to_expiry"][0]),
float(arrays["volatility"][0]),
option_type,
float(arrays["carry"][0]),
)
)
out = _rust_american_price_batch(
arrays["underlying"],
arrays["strike"],
arrays["rate"],
arrays["time_to_expiry"],
arrays["volatility"],
option_type,
arrays["carry"],
)
return np.asarray(out, dtype=np.float64)
except ValueError as err:
_normalize_rust_error(err)
def early_exercise_premium(
underlying: ArrayLike | float,
strike: ArrayLike | float,
rate: ArrayLike | float,
time_to_expiry: ArrayLike | float,
volatility: ArrayLike | float,
*,
option_type: str = "call",
carry: ArrayLike | float = 0.0,
) -> ScalarOrArray:
"""Early exercise premium: American price − European BSM price.
Represents the additional value an American option holder gains from the
right to exercise before expiry. Always ≥ 0.
Parameters
----------
underlying, strike, rate, time_to_expiry, volatility, option_type, carry:
Same as :func:`american_option_price`.
Returns
-------
float or NDArray[float64]
Premium ≥ 0. Typically 0 for calls with no dividends.
Notes
-----
For equity calls with zero carry (no dividends), early exercise is never
optimal so the premium is ≈ 0. For puts (or calls on dividend-paying
underlyings), the premium increases with in-the-moneyness, rate, and
time to expiry.
"""
from ferro_ta._ferro_ta import (
early_exercise_premium as _rust_early_exercise_premium,
)
from ferro_ta._ferro_ta import (
early_exercise_premium_batch as _rust_early_exercise_premium_batch,
)
option_type = _validate_option_type(option_type)
arrays, scalar_mode = _broadcast_inputs(
underlying=underlying,
strike=strike,
rate=rate,
time_to_expiry=time_to_expiry,
volatility=volatility,
carry=carry,
)
try:
if scalar_mode:
return float(
_rust_early_exercise_premium(
float(arrays["underlying"][0]),
float(arrays["strike"][0]),
float(arrays["rate"][0]),
float(arrays["time_to_expiry"][0]),
float(arrays["volatility"][0]),
option_type,
float(arrays["carry"][0]),
)
)
out = _rust_early_exercise_premium_batch(
arrays["underlying"],
arrays["strike"],
arrays["rate"],
arrays["time_to_expiry"],
arrays["volatility"],
option_type,
arrays["carry"],
)
return np.asarray(out, dtype=np.float64)
except ValueError as err:
_normalize_rust_error(err)
# ---------------------------------------------------------------------------
# Historical volatility estimators — populated once the Rust bridge is built
# ---------------------------------------------------------------------------
def close_to_close_vol(
close: ArrayLike,
window: int = 20,
trading_days_per_year: float = 252.0,
) -> NDArray[np.float64]:
"""Rolling close-to-close realized volatility (annualised).
Baseline estimator — uses only closing prices. Less efficient than OHLC
estimators but requires only daily close data.
Parameters
----------
close:
Array of closing prices (length ≥ window + 1).
window:
Rolling look-back period in bars (default 20).
trading_days_per_year:
Annualisation factor (default 252).
Returns
-------
NDArray[float64]
Same length as *close*. First ``window`` values are NaN.
Notes
-----
Formula::
σ = √( Σᵢ ln²(Cᵢ/Cᵢ₋₁) / window × trading_days_per_year )
No Bessel correction is applied (population variance, not sample variance).
"""
from ferro_ta._ferro_ta import close_to_close_vol as _rust_ctc
try:
arr = _to_f64(close)
return np.asarray(
_rust_ctc(arr, int(window), float(trading_days_per_year)), dtype=np.float64
)
except ValueError as err:
_normalize_rust_error(err)
def parkinson_vol(
high: ArrayLike,
low: ArrayLike,
window: int = 20,
trading_days_per_year: float = 252.0,
) -> NDArray[np.float64]:
"""Rolling Parkinson high-low realized volatility estimator (annualised).
~5× more efficient than close-to-close for diffusion processes.
Does **not** account for drift or overnight gaps.
Parameters
----------
high, low:
Arrays of daily high and low prices (same length, ≥ window).
window:
Rolling look-back period in bars (default 20).
trading_days_per_year:
Annualisation factor (default 252).
Returns
-------
NDArray[float64]
Same length as *high*. First ``window - 1`` values are NaN.
Notes
-----
Formula per window::
σ² = (1 / (4·ln2·window)) · Σ ln²(Hᵢ/Lᵢ) × trading_days_per_year
Reference: Parkinson, M. (1980). "The Extreme Value Method for
Estimating the Variance of the Rate of Return." *Journal of Business*, 53(1).
"""
from ferro_ta._ferro_ta import parkinson_vol as _rust_parkinson
try:
return np.asarray(
_rust_parkinson(
_to_f64(high), _to_f64(low), int(window), float(trading_days_per_year)
),
dtype=np.float64,
)
except ValueError as err:
_normalize_rust_error(err)
def garman_klass_vol(
open: ArrayLike,
high: ArrayLike,
low: ArrayLike,
close: ArrayLike,
window: int = 20,
trading_days_per_year: float = 252.0,
) -> NDArray[np.float64]:
"""Rolling Garman-Klass OHLC realized volatility estimator (annualised).
Extends Parkinson by incorporating the open-close return. ~7.4× more
efficient than close-to-close. Does **not** handle overnight gaps.
Parameters
----------
open, high, low, close:
Arrays of daily OHLC prices (same length, ≥ window).
window:
Rolling look-back period in bars (default 20).
trading_days_per_year:
Annualisation factor (default 252).
Returns
-------
NDArray[float64]
Same length as *close*. First ``window - 1`` values are NaN.
Notes
-----
Per-bar contribution::
GK = 0.5·ln²(H/L) − (2·ln2 − 1)·ln²(C/O)
Reference: Garman, M.B. & Klass, M.J. (1980). "On the Estimation of
Security Price Volatilities from Historical Data." *Journal of Business*, 53(1).
"""
from ferro_ta._ferro_ta import garman_klass_vol as _rust_gk
try:
return np.asarray(
_rust_gk(
_to_f64(open),
_to_f64(high),
_to_f64(low),
_to_f64(close),
int(window),
float(trading_days_per_year),
),
dtype=np.float64,
)
except ValueError as err:
_normalize_rust_error(err)
def rogers_satchell_vol(
open: ArrayLike,
high: ArrayLike,
low: ArrayLike,
close: ArrayLike,
window: int = 20,
trading_days_per_year: float = 252.0,
) -> NDArray[np.float64]:
"""Rolling Rogers-Satchell OHLC realized volatility estimator (annualised).
Drift-invariant: unbiased for assets with non-zero expected return.
Does **not** handle overnight gaps.
Parameters
----------
open, high, low, close:
Arrays of daily OHLC prices (same length, ≥ window).
window:
Rolling look-back period in bars (default 20).
trading_days_per_year:
Annualisation factor (default 252).
Returns
-------
NDArray[float64]
Same length as *close*. First ``window - 1`` values are NaN.
Notes
-----
Per-bar contribution (u = ln(H/O), d = ln(L/O), c = ln(C/O))::
RS = u·(u − c) + d·(d − c)
Reference: Rogers, L.C.G. & Satchell, S.E. (1991). "Estimating Variance
from High, Low and Closing Prices." *Annals of Applied Probability*, 1(4).
"""
from ferro_ta._ferro_ta import rogers_satchell_vol as _rust_rs
try:
return np.asarray(
_rust_rs(
_to_f64(open),
_to_f64(high),
_to_f64(low),
_to_f64(close),
int(window),
float(trading_days_per_year),
),
dtype=np.float64,
)
except ValueError as err:
_normalize_rust_error(err)
def yang_zhang_vol(
open: ArrayLike,
high: ArrayLike,
low: ArrayLike,
close: ArrayLike,
window: int = 20,
trading_days_per_year: float = 252.0,
) -> NDArray[np.float64]:
"""Rolling Yang-Zhang OHLC realized volatility estimator (annualised).
The most efficient standard estimator (~14× vs close-to-close). Handles
overnight gaps by combining overnight, intraday open-close, and
Rogers-Satchell variance components with an optimal weight *k*.
Parameters
----------
open, high, low, close:
Arrays of daily OHLC prices (same length, ≥ window + 1).
window:
Rolling look-back period in bars (default 20).
trading_days_per_year:
Annualisation factor (default 252).
Returns
-------
NDArray[float64]
Same length as *close*. First ``window`` values are NaN.
Notes
-----
Mixed estimator::
σ²_YZ = σ²_overnight + k·σ²_open_close + (1−k)·σ²_RS
where k = 0.34 / (1.34 + (window+1)/(window-1)).
Reference: Yang, D. & Zhang, Q. (2000). "Drift-Independent Volatility
Estimation Based on High, Low, Open, and Close Prices."
*Journal of Business*, 73(3).
"""
from ferro_ta._ferro_ta import yang_zhang_vol as _rust_yz
try:
return np.asarray(
_rust_yz(
_to_f64(open),
_to_f64(high),
_to_f64(low),
_to_f64(close),
int(window),
float(trading_days_per_year),
),
dtype=np.float64,
)
except ValueError as err:
_normalize_rust_error(err)
def vol_cone(
close: ArrayLike,
*,
windows: tuple[int, ...] = (21, 42, 63, 126, 252),
trading_days_per_year: float = 252.0,
) -> VolCone:
"""Historical realised vol distribution across window lengths (volatility cone).
For each window, computes the full history of rolling close-to-close
realised vol, then returns the min / p25 / median / p75 / max distribution.
Contextualises current implied vol: "Is 30 % IV cheap or expensive?"
Parameters
----------
close:
Array of closing prices (length ≥ max(windows) + 1).
windows:
Tuple of rolling window sizes in bars. Default ``(21, 42, 63, 126, 252)``
(approx. 1 month, 2 months, 3 months, 6 months, 1 year).
trading_days_per_year:
Annualisation factor (default 252).
Returns
-------
VolCone
Dataclass with arrays ``windows``, ``min``, ``p25``, ``median``,
``p75``, ``max`` — one value per element of *windows*.
Notes
-----
Uses close-to-close vol internally. Overlay the current IV on the cone
to see whether it is historically cheap or expensive for each tenor.
Examples
--------
>>> import numpy as np
>>> from ferro_ta.analysis.options import vol_cone
>>> rng = np.random.default_rng(0)
>>> close = 100 * np.cumprod(np.exp(rng.normal(0, 0.01, 500)))
>>> cone = vol_cone(close, windows=(21, 63, 252))
>>> cone.median # annualised median realised vol per window
"""
from ferro_ta._ferro_ta import vol_cone as _rust_vol_cone
try:
arr = _to_f64(close)
slices = _rust_vol_cone(arr, list(windows), float(trading_days_per_year))
windows_arr = np.array([s[0] for s in slices], dtype=np.float64)
return VolCone(
windows=windows_arr,
min=np.array([s[1] for s in slices], dtype=np.float64),
p25=np.array([s[2] for s in slices], dtype=np.float64),
median=np.array([s[3] for s in slices], dtype=np.float64),
p75=np.array([s[4] for s in slices], dtype=np.float64),
max=np.array([s[5] for s in slices], dtype=np.float64),
)
except ValueError as err:
_normalize_rust_error(err)
python/ferro_ta/analysis/options_strategy.py
+16 -7
View File
@@ -147,9 +147,9 @@ class SimulationLimits:
@dataclass(frozen=True)
class StrategyLeg:
underlying: str
expiry_selector: ExpirySelector
strike_selector: StrikeSelector
option_type: str
expiry_selector: ExpirySelector | None
strike_selector: StrikeSelector | None
option_type: str | None
side: str = "long"
quantity: int = 1
instrument: str = "option"
@@ -158,12 +158,21 @@ class StrategyLeg:
def __post_init__(self) -> None:
if self.underlying.strip() == "":
raise FerroTAInputError("underlying must not be empty.")
if self.option_type not in {"call", "put"}:
raise FerroTAValueError("option_type must be 'call' or 'put'.")
if self.instrument not in {"option", "future", "stock"}:
raise FerroTAValueError(
"instrument must be 'option', 'future', or 'stock'."
)
if self.instrument == "option":
if self.option_type not in {"call", "put"}:
raise FerroTAValueError(
"option legs require option_type='call' or 'put'."
)
if self.expiry_selector is None:
raise FerroTAInputError("option legs require expiry_selector.")
if self.strike_selector is None:
raise FerroTAInputError("option legs require strike_selector.")
if self.side not in {"long", "short"}:
raise FerroTAValueError("side must be 'long' or 'short'.")
if self.instrument not in {"option", "future"}:
raise FerroTAValueError("instrument must be 'option' or 'future'.")
if self.quantity == 0:
raise FerroTAValueError("quantity must be non-zero.")
if self.premium_limit is not None and self.premium_limit < 0.0: