-
Notifications
You must be signed in to change notification settings - Fork 0
/
volatility_report.py
333 lines (287 loc) · 12.3 KB
/
volatility_report.py
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
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
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
import os
import logging
from datetime import date, datetime, timedelta
import modal
from utils.options_watchlist import add_to_watchlist
logging.basicConfig(level=logging.INFO)
app = modal.App("volatility-analysis")
tastytrade_image = modal.Image.debian_slim(python_version="3.12").run_commands(
["pip install --upgrade tastytrade fastapi pydantic pytz"]
)
tickers_dict = modal.Dict.from_name("tickers-data", create_if_missing=True)
tastytrade_dict = modal.Dict.from_name("tastytrade-data", create_if_missing=True)
def get_tastytrade_session():
from tastytrade import Session
try:
if (
"session" in tastytrade_dict
and "session_created_at" in tastytrade_dict
and datetime.now() - tastytrade_dict["session_created_at"]
< timedelta(days=1)
):
logging.info("Tastytrade session is still valid.")
return tastytrade_dict["session"]
except KeyError:
logging.info("Creating new tastytrade session.")
try:
tastytrade_dict["session"] = Session(
os.environ["TASTYTRADE_USER"], os.environ["TASTYTRADE_PASSWORD"]
)
tastytrade_dict["session_created_at"] = datetime.now()
logging.info("Created new tastytrade session.")
except KeyError:
raise ValueError(
"TASTYTRADE_USER and/or TASTYTRADE_PASSWORD environment variables are not set."
)
return tastytrade_dict["session"]
def get_current_position_symbols(session) -> set[str]:
from tastytrade import Account
from tastytrade.utils import TastytradeError
try:
account = Account.get_account(session, os.getenv("TASTYTRADE_ACCOUNT"))
except TastytradeError:
accounts = Account.get_accounts(session)
if not accounts:
raise ValueError("No accounts found for the given session.")
account = accounts[0]
current_positions = account.get_positions(session)
# return the set of underlying symbols
return {position.underlying_symbol for position in current_positions}
def iv_rank_filter(
iv_rank: str | None, min_iv_rank: float = 0.2, max_iv_rank: float = 0.8
) -> bool:
return iv_rank is not None and min_iv_rank <= float(iv_rank) <= max_iv_rank
@app.function(
image=tastytrade_image,
secrets=[modal.Secret.from_name("tastytrade")],
)
def generate_report(required_tickers: list[str] = ()):
from decimal import Decimal
import pytz
import tastytrade
try:
if (
"volatility_report" in tastytrade_dict
and datetime.now() - tastytrade_dict["volatility_report_updated_at"]
< timedelta(minutes=10)
and all(ticker in tickers_dict["watchlist"] for ticker in required_tickers)
):
logging.info("Found cached volatility report.")
return tastytrade_dict["volatility_report"]
except KeyError:
logging.info(
"Error finding cached volatility report. Creating new volatility report."
)
try:
session = get_tastytrade_session()
current_position_symbols = get_current_position_symbols(session)
# ensure current positions and required tickers are in the watchlist:
add_to_watchlist(set(current_position_symbols) | set(required_tickers))
watchlist_metrics = tastytrade.metrics.get_market_metrics(
session, tickers_dict["watchlist"]
)
# Filter and sort metrics
filtered_metrics = [
x
for x in watchlist_metrics
if iv_rank_filter(x.implied_volatility_index_rank)
or x.symbol in current_position_symbols
or (
x.symbol in required_tickers
and x.implied_volatility_index_rank is not None
)
]
sorted_metrics = sorted(
filtered_metrics,
key=lambda x: float(x.implied_volatility_index_rank or 0),
reverse=True,
)
def calculate_last_updated(updated_at: datetime) -> str:
if not updated_at:
return "N/A"
metric_time = (
updated_at.replace(tzinfo=pytz.UTC)
if updated_at.tzinfo is None
else updated_at.astimezone(pytz.UTC)
)
time_difference = datetime.now(pytz.UTC) - metric_time
return (
f"{time_difference.days}d ago"
if time_difference.days
else f"{time_difference.seconds // 3600}h {(time_difference.seconds % 3600) // 60}m ago"
)
def calculate_days_to_earnings(
metric: tastytrade.metrics.MarketMetricInfo,
) -> str:
days_to_earnings = "N/A"
if metric.earnings and metric.earnings.expected_report_date:
if metric.earnings.expected_report_date >= date.today():
days_to_earnings = (
f"{(metric.earnings.expected_report_date - date.today()).days}"
)
return days_to_earnings
table_rows = []
for metric in sorted_metrics:
row = f"""
<tr>
<td>{metric.symbol + '*' if metric.symbol in current_position_symbols else metric.symbol}</td>
<td>{f"{Decimal(metric.implied_volatility_index_rank) * 100:.1f}%" if metric.implied_volatility_index_rank is not None else "N/A"}</td>
<td>{f"{Decimal(metric.implied_volatility_percentile) * 100:.1f}%" if metric.implied_volatility_percentile is not None else "N/A"}</td>
<td>{calculate_last_updated(metric.implied_volatility_updated_at)}</td>
<td>{f"{Decimal(metric.liquidity_rank) * 100:.1f}%" if metric.liquidity_rank is not None else "N/A"}</td>
<td>{metric.liquidity_rating if metric.liquidity_rating is not None else "N/A"}</td>
<td>{metric.lendability if metric.lendability is not None else "N/A"}</td>
<td>{f"{Decimal(metric.borrow_rate) * 100:.1f}%" if metric.borrow_rate is not None else "N/A"}</td>
<td>{calculate_days_to_earnings(metric)}</td>
</tr>
"""
table_rows.append(row)
headers = [
"Symbol",
"IV Rank",
"IV Percentile",
"Last Updated",
"Liquidity Rank",
"Liquidity Rating",
"Lendability",
"Borrow Rate",
"Days to Earnings",
]
header_row = "".join(f"<th>{header}</th>" for header in headers)
report = f"""
<table class="volatility-table">
<thead>
<tr>{header_row}</tr>
</thead>
<tbody>
{"".join(table_rows)}
</tbody>
</table>
"""
tastytrade_dict["volatility_report"] = report
tastytrade_dict["volatility_report_updated_at"] = datetime.now()
return tastytrade_dict["volatility_report"]
except Exception as e:
logging.error(f"Error generating volatility report: {e}")
raise e
@app.function(
allow_concurrent_inputs=10,
timeout=60,
)
@modal.asgi_app()
def serve():
from fastapi import FastAPI, Form
from fastapi.responses import HTMLResponse
app = FastAPI()
def generate_html_content(report: str, tickers: str = "") -> str:
return f"""
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Option Analysis</title>
<style>
body {{ font-family: Arial, sans-serif; margin: 0; padding: 20px; background-color: #121212; color: #e0e0e0; }}
h1 {{ color: #ffffff; }}
form {{ margin-bottom: 20px; }}
input[type="text"] {{ padding: 5px; width: 200px; background-color: #333; color: #e0e0e0; border: 1px solid #555; }}
input[type="submit"] {{ padding: 5px 10px; background-color: #4CAF50; color: white; border: none; cursor: pointer; }}
.table-container {{
max-height: 80vh;
overflow-y: auto;
}}
.volatility-table {{
width: 100%;
border-collapse: collapse;
background-color: #1e1e1e;
}}
.volatility-table th, .volatility-table td {{
border: 1px solid #333;
padding: 8px;
text-align: left;
}}
.volatility-table tr:nth-child(even) {{
background-color: #252525;
}}
.volatility-table thead {{
position: sticky;
top: 0;
background-color: #333;
color: white;
}}
.volatility-table th {{
background-color: #333;
color: white;
}}
.iv-rank-low {{ color: #4CAF50; }}
.iv-rank-high {{ color: #ff4444; }}
.iv-rank-very-high {{ color: #ff0000; }}
.liquidity-rating-low {{ color: #ff4444; }}
.lendability-locate {{ color: #ff4444; }}
.borrow-rate-high {{ color: #ff4444; }}
.borrow-rate-very-high {{ color: #ff0000; }}
</style>
</head>
<body>
<h1>Option Analysis</h1>
<form action="/refresh" method="post">
<input type="text" name="tickers" placeholder="Enter tickers (e.g., QQQ, SPY)" value="{tickers}">
<input type="submit" value="Refresh Report">
</form>
<div class="table-container">
{report}
</div>
<script>
document.querySelectorAll('.volatility-table td:nth-child(2)').forEach(cell => {{
const value = parseFloat(cell.textContent);
if (value <= 20) {{
cell.classList.add('iv-rank-low');
}} else if (value >= 80 && value < 100) {{
cell.classList.add('iv-rank-high');
}} else if (value >= 100) {{
cell.classList.add('iv-rank-very-high');
}}
}});
document.querySelectorAll('.volatility-table td:nth-child(3)').forEach(cell => {{
const value = parseFloat(cell.textContent);
if (value <= 20) {{
cell.classList.add('iv-rank-low');
}} else if (value >= 80) {{
cell.classList.add('iv-rank-high');
}}
}});
document.querySelectorAll('.volatility-table td:nth-child(6)').forEach(cell => {{
if (cell.textContent.trim() === '1') {{
cell.classList.add('liquidity-rating-low');
}}
}});
document.querySelectorAll('.volatility-table td:nth-child(7)').forEach(cell => {{
if (cell.textContent.trim() === 'Locate Required') {{
cell.classList.add('lendability-locate');
}}
}});
document.querySelectorAll('.volatility-table td:nth-child(8)').forEach(cell => {{
const value = parseFloat(cell.textContent);
if (value >= 100 && value < 1000) {{
cell.classList.add('borrow-rate-high');
}} else if (value >= 1000) {{
cell.classList.add('borrow-rate-very-high');
}}
}});
</script>
</body>
</html>
"""
@app.get("/", response_class=HTMLResponse)
async def root():
report = generate_report.remote()
return HTMLResponse(content=generate_html_content(report))
@app.post("/refresh", response_class=HTMLResponse)
async def refresh(tickers: str = Form(default="")):
required_tickers = [
ticker.strip() for ticker in tickers.split(",") if ticker.strip()
]
report = generate_report.remote(required_tickers)
return HTMLResponse(content=generate_html_content(report, tickers))
return app