Spaces:
Runtime error
Runtime error
File size: 5,225 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 | from __future__ import annotations
import json
from smolagents import tool
from market_data.providers import get_price_history
from strategy.payoff import expiration_payoff, strategy_summary
from strategy.schemas import OptionLeg, OptionStrategy
from .option_backtest import backtest_long_straddle_from_quotes, load_option_quotes_csv
from .vol_backtest import backtest_realized_vol_signal
def parse_legs(legs_json: str) -> list[OptionLeg]:
payload = json.loads(legs_json)
if isinstance(payload, dict) and "legs" in payload:
payload = payload["legs"]
return [OptionLeg(**leg) for leg in payload]
@tool
def analyze_strategy_payoff(legs_json: str, min_price: float, max_price: float, steps: int = 25) -> str:
"""Analyze expiration payoff for an option strategy.
Args:
legs_json: JSON list of option legs from build_volatility_strategy.
min_price: Minimum underlying price scenario.
max_price: Maximum underlying price scenario.
steps: Number of scenario steps.
"""
try:
legs = parse_legs(legs_json)
points = [
min_price + (max_price - min_price) * index / max(steps, 1)
for index in range(max(steps, 1) + 1)
]
rows = [
{"underlying_price": round(price, 2), "pnl": round(expiration_payoff(legs, price), 2)}
for price in points
]
temp_strategy = OptionStrategy(
name="custom_strategy",
volatility_view="unknown",
directional_view="unknown",
legs=legs,
rationale="custom payoff analysis",
risks=[],
max_profit=None,
max_loss=None,
breakevens=[],
net_debit_or_credit=round(sum(leg.premium * leg.signed_quantity() * 100 for leg in legs), 2),
score=0.0,
)
return json.dumps(
{
"status": "success",
"payoff_rows": rows,
"payoff_summary": strategy_summary(temp_strategy),
},
ensure_ascii=False,
indent=2,
)
except Exception as exc:
return json.dumps({"status": "error", "message": str(exc)}, ensure_ascii=False, indent=2)
@tool
def backtest_volatility_signal(
symbol: str,
signal: str = "long_vol",
period: str = "2y",
short_window: int = 10,
long_window: int = 30,
holding_days: int = 5,
) -> str:
"""Backtest a simple realized-volatility expansion/compression signal on the underlying.
Args:
symbol: Yahoo Finance ticker.
signal: long_vol or short_vol.
period: Yahoo Finance history period.
short_window: Short realized volatility lookback.
long_window: Long realized volatility lookback.
holding_days: Holding period after entry.
"""
try:
history = get_price_history(symbol, period=period, interval="1d")
result = backtest_realized_vol_signal(
history["Close"],
short_window=short_window,
long_window=long_window,
holding_days=holding_days,
signal=signal,
)
return json.dumps({"status": "success", "symbol": symbol.upper(), **result}, ensure_ascii=False, indent=2)
except Exception as exc:
return json.dumps({"status": "error", "symbol": symbol, "message": str(exc)}, ensure_ascii=False, indent=2)
@tool
def backtest_long_straddle_csv(
csv_path: str,
symbol: str,
target_dte: int = 30,
holding_days: int = 5,
entry_every_days: int = 5,
price_field: str = "trade",
) -> str:
"""Run a real option-quote backtest for repeated ATM long straddles.
This is a true option PnL backtest when supplied with historical option quotes.
Required CSV columns: date, underlying_symbol, underlying_price, contract_symbol,
option_type, expiration, strike, bid, ask. Optional columns include mid, delta,
gamma, theta, vega, implied_volatility, volume, open_interest.
Args:
csv_path: Path to historical option quotes CSV.
symbol: Underlying ticker.
target_dte: Target days to expiration at entry.
holding_days: Number of calendar days to hold each straddle.
entry_every_days: Minimum days between new entries.
price_field: trade for buy-at-ask/sell-at-bid, or mid for mid-price marks.
"""
try:
quotes = load_option_quotes_csv(csv_path)
result = backtest_long_straddle_from_quotes(
quotes=quotes,
symbol=symbol,
target_dte=target_dte,
holding_days=holding_days,
entry_every_days=entry_every_days,
price_field=price_field,
)
return json.dumps({"status": "success", **result}, ensure_ascii=False, indent=2)
except Exception as exc:
return json.dumps(
{
"status": "error",
"symbol": symbol,
"message": str(exc),
"note": "A real option backtest requires historical option quote data. yfinance does not provide reliable historical option chains.",
},
ensure_ascii=False,
indent=2,
)
|