Spaces:
Runtime error
Runtime error
File size: 7,780 Bytes
8f1601b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 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 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 | from __future__ import annotations
import json
from smolagents import tool
from .analytics import (
black_scholes_greeks,
classify_volatility_regime,
rank_current_iv_against_rv,
realized_volatility,
summarize_option_chain,
)
from .providers import get_current_quote, get_option_chain, get_price_history, list_option_expirations
from .schemas import VolSnapshot
def json_dumps(payload) -> str:
return json.dumps(payload, ensure_ascii=False, indent=2, default=str)
@tool
def query_market_asset(symbol: str) -> str:
"""Query the current price and intraday quote data for an asset.
Args:
symbol: Yahoo Finance ticker, e.g. AAPL, SPY, ^VIX, BTC-USD, EURUSD=X.
"""
try:
return json_dumps({"status": "success", **get_current_quote(symbol).to_dict()})
except Exception as exc:
return json_dumps({"status": "error", "symbol": symbol, "message": str(exc)})
@tool
def query_price_history(symbol: str, period: str = "1y", interval: str = "1d") -> str:
"""Query historical OHLCV prices for an asset.
Args:
symbol: Yahoo Finance ticker.
period: Yahoo Finance period such as 1mo, 6mo, 1y, 5y.
interval: Yahoo Finance interval such as 1d, 1h, 15m.
"""
try:
history = get_price_history(symbol, period=period, interval=interval)
records = history.tail(20).reset_index().to_dict(orient="records")
return json_dumps(
{
"status": "success",
"symbol": symbol.upper(),
"period": period,
"interval": interval,
"rows_returned": len(records),
"latest_rows": records,
}
)
except Exception as exc:
return json_dumps({"status": "error", "symbol": symbol, "message": str(exc)})
@tool
def query_realized_volatility(symbol: str, period: str = "1y") -> str:
"""Calculate realized volatility windows from historical close prices.
Args:
symbol: Yahoo Finance ticker.
period: Yahoo Finance history period.
"""
try:
history = get_price_history(symbol, period=period, interval="1d")
rv = realized_volatility(history["Close"])
return json_dumps({"status": "success", "symbol": symbol.upper(), "realized_volatility": rv})
except Exception as exc:
return json_dumps({"status": "error", "symbol": symbol, "message": str(exc)})
@tool
def query_option_expirations(symbol: str) -> str:
"""List available option expiration dates for an underlying.
Args:
symbol: Yahoo Finance ticker.
"""
try:
expirations = list_option_expirations(symbol)
return json_dumps({"status": "success", "symbol": symbol.upper(), "expirations": expirations})
except Exception as exc:
return json_dumps({"status": "error", "symbol": symbol, "message": str(exc)})
@tool
def query_option_chain(symbol: str, expiration: str = "") -> str:
"""Query an option chain with liquidity warnings and implied volatility.
Args:
symbol: Yahoo Finance ticker.
expiration: Expiration date in YYYY-MM-DD. Leave empty to use the nearest expiration.
"""
try:
chain = get_option_chain(symbol, expiration or None)
summary = summarize_option_chain(chain)
payload = chain.to_dict()
payload["summary"] = summary
payload["calls"] = payload["calls"][:80]
payload["puts"] = payload["puts"][:80]
return json_dumps({"status": "success", **payload})
except Exception as exc:
return json_dumps({"status": "error", "symbol": symbol, "message": str(exc)})
@tool
def query_volatility_snapshot(symbol: str, max_expirations: int = 4, history_period: str = "1y") -> str:
"""Summarize realized volatility, ATM IV, IV-RV spread, skew, and term structure.
Args:
symbol: Yahoo Finance ticker.
max_expirations: Number of expirations to sample from the option chain.
history_period: Yahoo Finance history period for realized volatility.
"""
try:
symbol = symbol.strip().upper()
quote = get_current_quote(symbol)
history = get_price_history(symbol, period=history_period, interval="1d")
rv = realized_volatility(history["Close"])
rv_20d = rv.get("20d")
expirations = list_option_expirations(symbol)[:max_expirations]
atm_iv_by_expiration = {}
iv_rv_spread_by_expiration = {}
skew_by_expiration = {}
for expiration in expirations:
chain = get_option_chain(symbol, expiration)
summary = summarize_option_chain(chain, realized_vol_20d=rv_20d)
atm_iv_by_expiration[expiration] = summary["atm_iv"]
iv_rv_spread_by_expiration[expiration] = summary["iv_rv_spread_20d"]
skew_by_expiration[expiration] = summary["skew_put_minus_call"]
valid_term_ivs = [
value
for value in atm_iv_by_expiration.values()
if value is not None
]
current_atm_iv = valid_term_ivs[0] if valid_term_ivs else None
sampled_skews = [value for value in skew_by_expiration.values() if value is not None]
front_skew = sampled_skews[0] if sampled_skews else None
term_structure_slope = (
float(valid_term_ivs[-1] - valid_term_ivs[0])
if len(valid_term_ivs) >= 2
else None
)
regime = classify_volatility_regime(
current_iv=current_atm_iv,
realized_vol_20d=rv_20d,
term_structure_slope=term_structure_slope,
skew=front_skew,
)
snapshot = VolSnapshot(
symbol=symbol,
current_price=quote.current_price,
realized_volatility=rv,
atm_iv_by_expiration=atm_iv_by_expiration,
iv_rv_spread_by_expiration=iv_rv_spread_by_expiration,
term_structure_slope=term_structure_slope,
skew_by_expiration=skew_by_expiration,
)
return json_dumps(
{
"status": "success",
**snapshot.to_dict(),
"front_atm_iv": current_atm_iv,
"front_skew": front_skew,
"iv_vs_rv_rank_proxy": rank_current_iv_against_rv(current_atm_iv, rv),
"volatility_regime": regime,
"limitations": [
"IV rank/percentile is a proxy based on current ATM IV versus realized-volatility windows.",
"True historical IV rank requires historical option-chain data from a richer provider.",
],
}
)
except Exception as exc:
return json_dumps({"status": "error", "symbol": symbol, "message": str(exc)})
@tool
def calculate_option_greeks(
spot: float,
strike: float,
time_to_expiry: float,
volatility: float,
option_type: str = "call",
risk_free_rate: float = 0.0,
dividend_yield: float = 0.0,
) -> str:
"""Calculate Black-Scholes-Merton Greeks for a single option.
Args:
spot: Current underlying price.
strike: Option strike.
time_to_expiry: Time to expiration in years.
volatility: Annualized implied volatility as a decimal.
option_type: call or put.
risk_free_rate: Annualized risk-free rate as a decimal.
dividend_yield: Annualized dividend yield as a decimal.
"""
greeks = black_scholes_greeks(
spot=spot,
strike=strike,
time_to_expiry=time_to_expiry,
volatility=volatility,
risk_free_rate=risk_free_rate,
dividend_yield=dividend_yield,
option_type=option_type,
)
return json_dumps({"status": "success", "greeks": greeks})
|