Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 28 additions & 6 deletions src/mcp_massive/constants.py
Original file line number Diff line number Diff line change
Expand Up @@ -90,6 +90,12 @@
"theta": ["bs_theta", "options"],
"vega": ["bs_vega", "options"],
"rho": ["bs_rho", "options"],
"vanna": ["bs_vanna", "options"],
"volga": ["bs_volga", "options"],
"vomma": ["bs_volga", "options"],
"charm": ["bs_charm", "options"],
"veta": ["bs_veta", "options"],
"color": ["bs_color", "options"],
"blackscholes": ["bs_price", "bs_delta"],
# technical indicators
"sma": "aggregate",
Expand Down Expand Up @@ -181,18 +187,34 @@
"Forex": {"forex", "fx", "currency", "currencies"},
"Options": {"option", "options", "call", "put", "strike", "chain"},
"Futures": {
"future", "futures", "futs",
"commodity", "commodities",
"crude", "oil", "gold", "silver", "gas", "wheat", "corn",
"cme", "nymex", "cbot",
"future",
"futures",
"futs",
"commodity",
"commodities",
"crude",
"oil",
"gold",
"silver",
"gas",
"wheat",
"corn",
"cme",
"nymex",
"cbot",
},
# NOTE: "index" deliberately omitted — collides with "consumer price
# index", "SEC EDGAR index", "filings index", etc. where the user
# means a table/document. Users who mean the asset class typically
# say "indices", "benchmark", or a specific index symbol.
"Indices": {
"indices", "benchmark",
"spx", "djia", "dow", "nikkei", "ftse",
"indices",
"benchmark",
"spx",
"djia",
"dow",
"nikkei",
"ftse",
},
"Economy": {"economy", "economic", "treasury", "inflation", "yield", "bond"},
}
Expand Down
182 changes: 145 additions & 37 deletions src/mcp_massive/functions.py
Original file line number Diff line number Diff line change
Expand Up @@ -205,18 +205,30 @@ def _bs_d1d2(
return d1, d2


def _bs_inputs(
inputs: dict[str, Any], n: int
) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
"""Resolve the five common BS params (S, K, T, r, sigma) to numpy arrays.

apply_pipeline guarantees these keys are present before the impl runs;
no defensive checks needed here.
"""
return (
_to_numpy(inputs["S"], n),
_to_numpy(inputs["K"], n),
_to_numpy(inputs["T"], n),
_to_numpy(inputs["r"], n),
_to_numpy(inputs["sigma"], n),
)


# ---------------------------------------------------------------------------
# Greeks implementations
# ---------------------------------------------------------------------------


def _impl_bs_price(table: Table, inputs: dict[str, Any]) -> np.ndarray:
n = len(table)
S = _to_numpy(inputs["S"], n)
K = _to_numpy(inputs["K"], n)
T = _to_numpy(inputs["T"], n)
r = _to_numpy(inputs["r"], n)
sigma = _to_numpy(inputs["sigma"], n)
S, K, T, r, sigma = _bs_inputs(inputs, len(table))
option_type = _validate_option_type(inputs)

d1, d2 = _bs_d1d2(S, K, T, r, sigma)
Expand All @@ -228,12 +240,7 @@ def _impl_bs_price(table: Table, inputs: dict[str, Any]) -> np.ndarray:


def _impl_bs_delta(table: Table, inputs: dict[str, Any]) -> np.ndarray:
n = len(table)
S = _to_numpy(inputs["S"], n)
K = _to_numpy(inputs["K"], n)
T = _to_numpy(inputs["T"], n)
r = _to_numpy(inputs["r"], n)
sigma = _to_numpy(inputs["sigma"], n)
S, K, T, r, sigma = _bs_inputs(inputs, len(table))
option_type = _validate_option_type(inputs)

d1, _d2 = _bs_d1d2(S, K, T, r, sigma)
Expand All @@ -245,25 +252,15 @@ def _impl_bs_delta(table: Table, inputs: dict[str, Any]) -> np.ndarray:


def _impl_bs_gamma(table: Table, inputs: dict[str, Any]) -> np.ndarray:
n = len(table)
S = _to_numpy(inputs["S"], n)
K = _to_numpy(inputs["K"], n)
T = _to_numpy(inputs["T"], n)
r = _to_numpy(inputs["r"], n)
sigma = _to_numpy(inputs["sigma"], n)
S, K, T, r, sigma = _bs_inputs(inputs, len(table))

d1, _d2 = _bs_d1d2(S, K, T, r, sigma)
gamma = _norm_pdf(d1) / (S * sigma * np.sqrt(T))
return gamma


def _impl_bs_theta(table: Table, inputs: dict[str, Any]) -> np.ndarray:
n = len(table)
S = _to_numpy(inputs["S"], n)
K = _to_numpy(inputs["K"], n)
T = _to_numpy(inputs["T"], n)
r = _to_numpy(inputs["r"], n)
sigma = _to_numpy(inputs["sigma"], n)
S, K, T, r, sigma = _bs_inputs(inputs, len(table))
option_type = _validate_option_type(inputs)

d1, d2 = _bs_d1d2(S, K, T, r, sigma)
Expand All @@ -277,12 +274,7 @@ def _impl_bs_theta(table: Table, inputs: dict[str, Any]) -> np.ndarray:


def _impl_bs_vega(table: Table, inputs: dict[str, Any]) -> np.ndarray:
n = len(table)
S = _to_numpy(inputs["S"], n)
K = _to_numpy(inputs["K"], n)
T = _to_numpy(inputs["T"], n)
r = _to_numpy(inputs["r"], n)
sigma = _to_numpy(inputs["sigma"], n)
S, K, T, r, sigma = _bs_inputs(inputs, len(table))

d1, _d2 = _bs_d1d2(S, K, T, r, sigma)
vega = S * _norm_pdf(d1) * np.sqrt(T)
Expand All @@ -291,12 +283,7 @@ def _impl_bs_vega(table: Table, inputs: dict[str, Any]) -> np.ndarray:


def _impl_bs_rho(table: Table, inputs: dict[str, Any]) -> np.ndarray:
n = len(table)
S = _to_numpy(inputs["S"], n)
K = _to_numpy(inputs["K"], n)
T = _to_numpy(inputs["T"], n)
r = _to_numpy(inputs["r"], n)
sigma = _to_numpy(inputs["sigma"], n)
S, K, T, r, sigma = _bs_inputs(inputs, len(table))
option_type = _validate_option_type(inputs)

_d1, d2 = _bs_d1d2(S, K, T, r, sigma)
Expand All @@ -308,6 +295,72 @@ def _impl_bs_rho(table: Table, inputs: dict[str, Any]) -> np.ndarray:
return rho / 100.0


def _impl_bs_vanna(table: Table, inputs: dict[str, Any]) -> np.ndarray:
S, K, T, r, sigma = _bs_inputs(inputs, len(table))

d1, d2 = _bs_d1d2(S, K, T, r, sigma)
vanna = -_norm_pdf(d1) * d2 / sigma
# Per 1% change in volatility (matches bs_vega scaling)
return vanna / 100.0


def _impl_bs_volga(table: Table, inputs: dict[str, Any]) -> np.ndarray:
S, K, T, r, sigma = _bs_inputs(inputs, len(table))

d1, d2 = _bs_d1d2(S, K, T, r, sigma)
vega = S * _norm_pdf(d1) * np.sqrt(T)
volga = vega * d1 * d2 / sigma
# Per (1% vol)^2 — change in bs_vega per 1% vol change
return volga / 10000.0


# Calendar-time time-decay Greeks: ∂x/∂t (bleeds with passage of time).
# Same sign convention as bs_theta — negative for the long-option holder
# in the typical ATM case. Assumes no dividends (q=0); under that
# assumption charm/veta/color are identical for calls and puts.


def _impl_bs_charm(table: Table, inputs: dict[str, Any]) -> np.ndarray:
S, K, T, r, sigma = _bs_inputs(inputs, len(table))

d1, d2 = _bs_d1d2(S, K, T, r, sigma)
sqrt_T = np.sqrt(T)
charm = (
-_norm_pdf(d1) * (2 * r * T - d2 * sigma * sqrt_T) / (2 * T * sigma * sqrt_T)
)
# Per day (matches bs_theta /365 convention)
return charm / 365.0


def _impl_bs_veta(table: Table, inputs: dict[str, Any]) -> np.ndarray:
S, K, T, r, sigma = _bs_inputs(inputs, len(table))

d1, d2 = _bs_d1d2(S, K, T, r, sigma)
sqrt_T = np.sqrt(T)
veta = (
S
* _norm_pdf(d1)
* sqrt_T
* (r * d1 / (sigma * sqrt_T) - (1.0 + d1 * d2) / (2.0 * T))
)
# Per day, per 1% vol — change in bs_vega per day
return veta / 36500.0


def _impl_bs_color(table: Table, inputs: dict[str, Any]) -> np.ndarray:
S, K, T, r, sigma = _bs_inputs(inputs, len(table))

d1, d2 = _bs_d1d2(S, K, T, r, sigma)
sqrt_T = np.sqrt(T)
color = (
_norm_pdf(d1)
/ (2.0 * S * T * sigma * sqrt_T)
* (1.0 + (2.0 * r * T - d2 * sigma * sqrt_T) * d1 / (sigma * sqrt_T))
)
# Per day — change in bs_gamma per day
return color / 365.0


# ---------------------------------------------------------------------------
# Returns implementations (numpy)
# ---------------------------------------------------------------------------
Expand Down Expand Up @@ -547,7 +600,7 @@ def _impl_ema(table: Table, inputs: dict[str, Any]) -> np.ndarray:
FunctionDef(
name="bs_theta",
category="Greeks",
description="Black-Scholes daily theta (annual theta / 365).",
description="Black-Scholes daily theta. Time decay — change in option price per day (annual theta / 365).",
params=[*_BS_COMMON_PARAMS, _BS_OPTION_TYPE_PARAM],
output_dtype="Float64",
impl=_impl_bs_theta,
Expand Down Expand Up @@ -576,6 +629,61 @@ def _impl_ema(table: Table, inputs: dict[str, Any]) -> np.ndarray:
)
)

_register(
FunctionDef(
name="bs_vanna",
category="Greeks",
description="Black-Scholes vanna (dDelta/dSigma = dVega/dSpot) per 1% change in volatility. Same for calls and puts.",
params=list(_BS_COMMON_PARAMS),
output_dtype="Float64",
impl=_impl_bs_vanna,
)
)

_register(
FunctionDef(
name="bs_volga",
category="Greeks",
description="Black-Scholes volga / vomma. Change in bs_vega per 1% volatility change. Same for calls and puts.",
params=list(_BS_COMMON_PARAMS),
output_dtype="Float64",
impl=_impl_bs_volga,
)
)

_register(
FunctionDef(
name="bs_charm",
category="Greeks",
description="Black-Scholes charm (delta decay). Change in bs_delta per day. Assumes no dividends; same for calls and puts under that assumption.",
params=list(_BS_COMMON_PARAMS),
output_dtype="Float64",
impl=_impl_bs_charm,
)
)

_register(
FunctionDef(
name="bs_veta",
category="Greeks",
description="Black-Scholes veta / DvegaDtime (vega decay). Change in bs_vega per day. Assumes no dividends; same for calls and puts under that assumption.",
params=list(_BS_COMMON_PARAMS),
output_dtype="Float64",
impl=_impl_bs_veta,
)
)

_register(
FunctionDef(
name="bs_color",
category="Greeks",
description="Black-Scholes color (gamma decay). Change in bs_gamma per day. Assumes no dividends; same for calls and puts under that assumption.",
params=list(_BS_COMMON_PARAMS),
output_dtype="Float64",
impl=_impl_bs_color,
)
)

_register(
FunctionDef(
name="simple_return",
Expand Down
Loading
Loading