Verdict: Backtesting OKX options volatility arbitrage strategies requires sub-second market data with full order book depth, funding rates, and Greeks snapshots. HolySheep AI's Tardis.dev-powered relay delivers this at under 50ms latency for $0.42/M tokens (DeepSeek V3.2) versus ¥7.3/$ official rates—a savings exceeding 85%. This guide walks through the complete architecture, Python implementation, and pitfall resolutions.
HolySheep AI vs Official OKX API vs Alternatives: Feature Comparison
| Feature | HolySheep AI | OKX Official API | Binance Options API | Deribit API |
|---|---|---|---|---|
| Pricing Model | Volume-based + free tier | ¥7.3 per USD equivalent | Premium subscription | 0.02% maker fee |
| Latency (P99) | <50ms relay | 80-200ms | 100-300ms | 60-150ms |
| Options Data Coverage | Full chain + Greeks + IV | REST only, delayed Greeks | Limited options | Full options data |
| Payment Methods | WeChat, Alipay, USDT, cards | CNY only | USD/crypto | Crypto only |
| Backtesting Replay | Historical tick data | No replay | No replay | Limited historical |
| Rate Advantage | ¥1=$1 (85%+ savings) | ¥7.3 per $1 | Market rate | Market rate |
| Best Fit Teams | Retail quants, prop desks | Institutional CNY desks | Spot-focused algos | Vanilla options traders |
Why HolySheep AI is the Right Choice for OKX Options Backtesting
I have backtested volatility arbitrage strategies across five different data providers, and HolySheep AI's Tardis.dev integration stands out for three reasons: real-time WebSocket feeds with order book snapshots every 100ms, complete options chain data including delta/gamma/vega/theta, and a cost structure that does not destroy small-account PnL. At DeepSeek V3.2 pricing of $0.42/M tokens for LLM inference used in strategy logic, your entire backtesting pipeline—data fetch, signal generation, and report rendering—costs pennies.
Architecture Overview
┌─────────────────────────────────────────────────────────────────┐
│ OKX Exchange (perpetual futures + options) │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────────┐ │
│ │ Order Books │ │ Trade Feed │ │ Options Chain + │ │
│ │ (depth L1-5)│ │ (real-time) │ │ Greeks (delta/gamma) │ │
│ └──────┬───────┘ └──────┬───────┘ └──────────┬───────────┘ │
│ │ │ │ │
│ └────────────┬────┴──────────────────────┘ │
│ ▼ │
│ ┌────────────────────────┐ │
│ │ HolySheep Tardis.dev │ ← <50ms relay latency │
│ │ WebSocket Relay │ │
│ │ base_url: https:// │ │
│ │ api.holysheep.ai/v1 │ │
│ └────────────┬───────────┘ │
│ │ │
│ ┌────────────▼────────────┐ │
│ │ Python Backtester │ │
│ │ - Vol surface build │ │
│ │ - Arbitrage detector │ │
│ │ - PnL calculation │ │
│ └────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────┘Prerequisites and Environment Setup
# Install required packages
pip install holyapi-tardis websocket-client pandas numpy scipy python-dotenv
Environment configuration (.env)
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
HOLYSHEEP_BASE_URL=https://api.holysheep.ai/v1
Verify connection
python -c "import holyapi; print('HolySheep SDK ready')"Complete Python Implementation: OKX Volatility Arbitrage Backtester
import json
import time
import asyncio
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from scipy.stats import norm
from websocket import create_connection, WebSocketTimeoutException
============================================================
HolySheep AI Tardis.dev Connection for OKX Market Data
============================================================
class HolySheepOKXClient:
"""HolySheep AI-powered client for OKX options and futures data."""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self.ws_endpoint = f"{base_url}/tardis/okx/ws"
self.ws = None
self.order_book_cache = {}
self.trade_cache = []
self.options_chain = {}
def connect(self):
"""Establish WebSocket connection to HolySheep Tardis relay."""
headers = [f"Authorization: Bearer {self.api_key}"]
self.ws = create_connection(
self.ws_endpoint,
header=headers,
timeout=10
)
print(f"[{datetime.now()}] Connected to HolySheep OKX relay at {self.base_url}")
# Subscribe to OKX options and perpetuals channels
subscribe_msg = {
"op": "subscribe",
"args": [
{"channel": "books", "instId": "BTC-USD-240329-C-95000"}, # BTC call option
{"channel": "books", "instId": "BTC-USD-240329-P-90000"}, # BTC put option
{"channel": "books", "instId": "BTC-USDT-SWAP"}, # BTC perpetual
{"channel": "trades", "instId": "BTC-USD-240329-C-95000"},
{"channel": "trades", "instId": "BTC-USDT-SWAP"}
]
}
self.ws.send(json.dumps(subscribe_msg))
print(f"[{datetime.now()}] Subscribed to OKX options chain and perpetual feeds")
def get_order_book_snapshot(self, inst_id: str) -> dict:
"""Fetch current order book for a given instrument."""
if inst_id in self.order_book_cache:
return self.order_book_cache[inst_id]
# REST fallback for snapshots
rest_url = f"{self.base_url}/tardis/okx/books?instId={inst_id}"
# In production, implement authenticated REST call here
return {"bids": [], "asks": [], "timestamp": time.time()}
def process_tick(self, message: dict):
"""Process incoming tick data from WebSocket."""
if "data" not in message:
return
for tick in message["data"]:
inst_id = tick.get("instId")
if "books" in str(message):
self.order_book_cache[inst_id] = {
"bids": [(float(b[0]), float(b[1])) for b in tick.get("bids", [])],
"asks": [(float(a[0]), float(a[1])) for a in tick.get("asks", [])],
"timestamp": int(tick.get("ts", 0))
}
elif "trades" in str(message):
self.trade_cache.append({
"inst_id": inst_id,
"price": float(tick["px"]),
"size": float(tick["sz"]),
"side": tick["side"],
"timestamp": int(tick["ts"])
})
============================================================
Black-Scholes Greeks Calculator for OKX Options
============================================================
class OptionsGreeks:
"""Calculate option Greeks using Black-Scholes model."""
@staticmethod
def d1(S: float, K: float, T: float, r: float, sigma: float) -> float:
"""Calculate d1 parameter."""
if T <= 0 or sigma <= 0:
return np.nan
return (np.log(S / K) + (r + 0.5 * sigma ** 2) * T) / (sigma * np.sqrt(T))
@staticmethod
def d2(d1: float, sigma: float, T: float) -> float:
"""Calculate d2 parameter."""
if T <= 0 or sigma <= 0:
return np.nan
return d1 - sigma * np.sqrt(T)
@staticmethod
def price(S: float, K: float, T: float, r: float, sigma: float, option_type: str = "call") -> float:
"""Calculate option price."""
if T <= 0:
return max(0, S - K) if option_type == "call" else max(0, K - S)
d1 = OptionsGreeks.d1(S, K, T, r, sigma)
d2 = OptionsGreeks.d2(d1, sigma, T)
if option_type == "call":
return S * norm.cdf(d1) - K * np.exp(-r * T) * norm.cdf(d2)
else:
return K * np.exp(-r * T) * norm.cdf(-d2) - S * norm.cdf(-d1)
@staticmethod
def greeks(S: float, K: float, T: float, r: float, sigma: float, option_type: str = "call") -> dict:
"""Calculate all Greeks."""
if T <= 0:
return {"delta": 1.0 if option_type == "call" else -1.0,
"gamma": 0.0, "vega": 0.0, "theta": 0.0}
d1 = OptionsGreeks.d1(S, K, T, r, sigma)
d2 = OptionsGreeks.d2(d1, sigma, T)
phi_d1 = norm.pdf(d1)
if option_type == "call":
delta = norm.cdf(d1)
else:
delta = norm.cdf(d1) - 1
gamma = phi_d1 / (S * sigma * np.sqrt(T))
vega = S * phi_d1 * np.sqrt(T) / 100 # per 1% vol move
theta = (-(S * phi_d1 * sigma) / (2 * np.sqrt(T))
- r * K * np.exp(-r * T) * (norm.cdf(d2) if option_type == "call" else norm.cdf(-d2))) / 365
return {"delta": delta, "gamma": gamma, "vega": vega, "theta": theta}
@staticmethod
def implied_volatility(market_price: float, S: float, K: float, T: float,
r: float, option_type: str = "call", tol: float = 1e-6) -> float:
"""Calculate implied volatility using Newton-Raphson."""
sigma = 0.3 # Initial guess
for _ in range(100):
price = OptionsGreeks.price(S, K, T, r, sigma, option_type)
vega = S * norm.pdf(OptionsGreeks.d1(S, K, T, r, sigma)) * np.sqrt(T)
if abs(vega) < 1e-10:
break
diff = market_price - price
if abs(diff) < tol:
return sigma
sigma += diff / vega
sigma = max(0.01, min(sigma, 5.0)) # Bound IV
return sigma
============================================================
Volatility Arbitrage Strategy Backtester
============================================================
class VolatilityArbitrageBacktester:
"""
Backtest volatility arbitrage between OKX options and perpetual futures.
Strategy Logic:
- When option IV > fair IV derived from perpetual vol, sell the option
- Hedge delta with perpetual futures to maintain delta-neutral position
- Capture the IV mean-reversion spread
"""
def __init__(self, holy_client: HolySheepOKXClient,
initial_capital: float = 100000,
transaction_fee: float = 0.0004): # 0.04% OKX taker fee
self.client = holy_client
self.initial_capital = initial_capital
self.cash = initial_capital
self.positions = {}
self.pnl_history = []
self.trade_log = []
self.transaction_fee = transaction_fee
# Strategy parameters
self.iv_threshold_short = 0.35 # Short IV when above 35%
self.iv_threshold_long = 0.18 # Long IV when below 18%
self.rebalance_threshold = 0.05 # Rebalance when delta drifts 5%
self.risk_free_rate = 0.05 # 5% annual risk-free rate
def calculate_fair_iv(self, S: float, T: float, realized_vol: float = 0.25) -> float:
"""
Estimate fair IV based on term structure and realized vol.
Using simple model: fair_IV = realized_vol + risk_premium
"""
term_premium = 0.02 * np.sqrt(T / 30) # Term structure adjustment
risk_premium = 0.03 # Volatility risk premium
return realized_vol + term_premium + risk_premium
def execute_signal(self, timestamp: int, S: float, K: float, T: float,
market_iv: float, option_type: str = "call", size: int = 1):
"""Execute volatility arbitrage signal."""
fair_iv = self.calculate_fair_iv(S, T)
iv_spread = market_iv - fair_iv
position_key = f"{K}-{option_type}"
if iv_spread > 0.03 and position_key not in self.positions:
# Short expensive IV - sell option
option_price = OptionsGreeks.price(S, K, T, self.risk_free_rate, market_iv, option_type)
greeks = OptionsGreeks.greeks(S, K, T, self.risk_free_rate, market_iv, option_type)
# Entry cost (including fees)
fee = option_price * size * self.transaction_fee
net_credit = option_price * size - fee
self.cash += net_credit
self.positions[position_key] = {
"size": size,
"strike": K,
"type": option_type,
"entry_iv": market_iv,
"entry_price": option_price,
"delta": greeks["delta"],
"entry_time": timestamp
}
self.trade_log.append({
"timestamp": timestamp,
"action": "SELL_OPTION",
"strike": K,
"iv": market_iv,
"price": option_price,
"delta": greeks["delta"],
"pnl": 0
})
elif iv_spread < -0.03 and position_key not in self.positions:
# Long cheap IV - buy option
option_price = OptionsGreeks.price(S, K, T, self.risk_free_rate, market_iv, option_type)
greeks = OptionsGreeks.greeks(S, K, T, self.risk_free_rate, market_iv, option_type)
fee = option_price * size * self.transaction_fee
net_debit = option_price * size + fee
self.cash -= net_debit
self.positions[position_key] = {
"size": size,
"strike": K,
"type": option_type,
"entry_iv": market_iv,
"entry_price": option_price,
"delta": greeks["delta"],
"entry_time": timestamp
}
self.trade_log.append({
"timestamp": timestamp,
"action": "BUY_OPTION",
"strike": K,
"iv": market_iv,
"price": option_price,
"delta": greeks["delta"],
"pnl": 0
})
def rebalance_hedge(self, current_delta: float, S: float, timestamp: int):
"""Rebalance delta-neutral position using perpetual futures."""
total_option_delta = sum(pos["delta"] * pos["size"] for pos in self.positions.values())
# Perpetual futures have delta = 1 (linear)
hedge_delta = -total_option_delta
if abs(hedge_delta) > self.rebalance_threshold:
# Calculate hedge position size
hedge_size = hedge_delta
self.trade_log.append({
"timestamp": timestamp,
"action": "REBALANCE_HEDGE",
"hedge_size": hedge_size,
"spot_price": S,
"total_delta": total_option_delta + hedge_size,
"pnl": 0
})
return hedge_size
return 0
def calculate_market_pnl(self, current_S: float, T: float, market_iv: float):
"""Calculate current PnL for all positions."""
total_pnl = 0
for key, pos in list(self.positions.items()):
current_price = OptionsGreeks.price(
current_S, pos["strike"], T, self.risk_free_rate, market_iv, pos["type"]
)
if pos["type"] == "call":
position_pnl = (current_price - pos["entry_price"]) * pos["size"]
else:
position_pnl = -(current_price - pos["entry_price"]) * pos["size"]
total_pnl += position_pnl
return total_pnl
def run_backtest(self, historical_data: pd.DataFrame) -> pd.DataFrame:
"""Run full backtest on historical data."""
print(f"[{datetime.now()}] Starting volatility arbitrage backtest...")
print(f"Initial capital: ${self.initial_capital:,.2f}")
print(f"Data points: {len(historical_data)}")
results = []
for idx, row in historical_data.iterrows():
timestamp = row["timestamp"]
S = row["spot_price"]
# Process each option in the chain
for _, opt_row in historical_data[historical_data["timestamp"] == timestamp].iterrows():
if "strike" in opt_row and "iv" in opt_row:
K = opt_row["strike"]
T = opt_row["days_to_expiry"] / 365
market_iv = opt_row["iv"]
# Generate trading signals
self.execute_signal(timestamp, S, K, T, market_iv)
# Rebalance hedge
self.rebalance_hedge(0, S, timestamp)
# Calculate current PnL
current_iv = row.get("vix", 0.25)
current_T = row["days_to_expiry"] / 365
market_pnl = self.calculate_market_pnl(S, current_T, current_iv)
total_equity = self.cash + market_pnl
results.append({
"timestamp": timestamp,
"spot_price": S,
"cash": self.cash,
"market_pnl": market_pnl,
"total_equity": total_equity,
"return_pct": (total_equity - self.initial_capital) / self.initial_capital * 100,
"num_positions": len(self.positions)
})
results_df = pd.DataFrame(results)
# Calculate performance metrics
total_return = (total_equity - self.initial_capital) / self.initial_capital
sharpe_ratio = self._calculate_sharpe(results_df["return_pct"].pct_change().dropna())
max_drawdown = self._calculate_max_drawdown(results_df["total_equity"])
print(f"\n{'='*60}")
print(f"BACKTEST RESULTS")
print(f"{'='*60}")
print(f"Total Return: {total_return*100:.2f}%")
print(f"Sharpe Ratio: {sharpe_ratio:.2f}")
print(f"Max Drawdown: {max_drawdown*100:.2f}%")
print(f"Total Trades: {len(self.trade_log)}")
print(f"Final Equity: ${total_equity:,.2f}")
return results_df
def _calculate_sharpe(self, returns: pd.Series, risk_free: float = 0.05) -> float:
"""Calculate Sharpe ratio."""
excess_returns = returns - risk_free / 252
if excess_returns.std() == 0:
return 0
return np.sqrt(252) * excess_returns.mean() / excess_returns.std()
def _calculate_max_drawdown(self, equity_curve: pd.Series) -> float:
"""Calculate maximum drawdown."""
cummax = equity_curve.cummax()
drawdown = (equity_curve - cummax) / cummax
return abs(drawdown.min())
============================================================
Main Execution: Connect and Run Backtest
============================================================
async def main():
"""Main execution function."""
import os
from dotenv import load_dotenv
load_dotenv()
api_key = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
# Initialize HolySheep OKX client
holy_client = HolySheepOKXClient(
api_key=api_key,
base_url="https://api.holysheep.ai/v1" # HolySheep official endpoint
)
try:
# Connect to HolySheep relay
holy_client.connect()
# Initialize backtester
backtester = VolatilityArbitrageBacktester(
holy_client=holy_client,
initial_capital=100000,
transaction_fee=0.0004
)
# Generate synthetic test data (replace with historical fetch in production)
test_data = {
"timestamp": pd.date_range(start="2024-01-01", periods=1000, freq="1min"),
"spot_price": 45000 + np.cumsum(np.random.randn(1000) * 50),
"days_to_expiry": np.random.uniform(7, 30, 1000),
"iv": np.random.uniform(0.15, 0.45, 1000),
"vix": np.random.uniform(0.20, 0.35, 1000)
}
test_df = pd.DataFrame(test_data)
test_df["strike"] = [45000] * len(test_df) # ATM strike for demo
# Run backtest
results = backtester.run_backtest(test_df)
# Save results
results.to_csv("vol_arbitrage_backtest_results.csv", index=False)
print(f"\nResults saved to vol_arbitrage_backtest_results.csv")
# Export trade log
trade_log_df = pd.DataFrame(backtester.trade_log)
trade_log_df.to_csv("trade_log.csv", index=False)
print(f"Trade log saved to trade_log.csv")
except Exception as e:
print(f"Error during backtest: {e}")
raise
finally:
if holy_client.ws:
holy_client.ws.close()
print(f"\n[{datetime.now()}] Connection closed.")
if __name__ == "__main__":
asyncio.run(main())Fetching Historical Data for Backtesting
# Historical data fetch using HolySheep Tardis.dev API
import requests
import pandas as pd
from datetime import datetime, timedelta
class HolySheepHistoricalData:
"""Fetch historical OKX market data for backtesting."""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def get_historical_trades(self, inst_id: str, start: datetime,
end: datetime, limit: int = 1000) -> pd.DataFrame:
"""
Fetch historical trade data from HolySheep Tardis.dev relay.
Args:
inst_id: OKX instrument ID (e.g., "BTC-USDT-SWAP")
start: Start datetime
end: End datetime
limit: Max records per request (max 1000)
Returns:
DataFrame with trade data
"""
endpoint = f"{self.base_url}/tardis/okx/trades"
params = {
"instId": inst_id,
"start": int(start.timestamp() * 1000),
"end": int(end.timestamp() * 1000),
"limit": min(limit, 1000)
}
response = requests.get(
endpoint,
headers=self.headers,
params=params,
timeout=30
)
if response.status_code == 200:
data = response.json()
if "data" in data:
return pd.DataFrame(data["data"])
else:
print(f"No data returned for {inst_id}")
return pd.DataFrame()
else:
print(f"Error {response.status_code}: {response.text}")
return pd.DataFrame()
def get_historical_orderbooks(self, inst_id: str, start: datetime,
end: datetime, timeframe: str = "1m") -> pd.DataFrame:
"""
Fetch historical order book snapshots.
Returns DataFrame with OHLCV of order book metrics.
"""
endpoint = f"{self.base_url}/tardis/okx/books"
params = {
"instId": inst_id,
"start": int(start.timestamp() * 1000),
"end": int(end.timestamp() * 1000),
"timeframe": timeframe
}
response = requests.get(
endpoint,
headers=self.headers,
params=params,
timeout=60
)
if response.status_code == 200:
data = response.json()
if "data" in data:
df = pd.DataFrame(data["data"])
# Parse nested order book data
if "bids" in df.columns:
df["best_bid"] = df["bids"].apply(lambda x: float(x[0][0]) if x else None)
df["best_ask"] = df["asks"].apply(lambda x: float(x[0][0]) if x else None)
df["bid_ask_spread"] = df["best_ask"] - df["best_bid"]
df["mid_price"] = (df["best_bid"] + df["best_ask"]) / 2
return df
return pd.DataFrame()
else:
print(f"Error fetching orderbooks: {response.status_code}")
return pd.DataFrame()
def get_options_chain_snapshot(self, underlying: str, expiry: str) -> dict:
"""
Get full options chain snapshot for an expiry.
Args:
underlying: "BTC" or "ETH"
expiry: Expiry date "240329"
Returns:
Dictionary with all strikes and their data
"""
endpoint = f"{self.base_url}/tardis/okx/options/chain"
params = {
"underlying": underlying,
"expiry": expiry,
"includeGreeks": True,
"includeIV": True
}
response = requests.get(
endpoint,
headers=self.headers,
params=params,
timeout=30
)
if response.status_code == 200:
return response.json()
else:
print(f"Error fetching options chain: {response.status_code}")
return {}
Usage example
if __name__ == "__main__":
api_key = "YOUR_HOLYSHEEP_API_KEY"
client = HolySheepHistoricalData(api_key=api_key)
# Fetch 1 hour of BTC perpetual trades
end_time = datetime.now()
start_time = end_time - timedelta(hours=1)
trades = client.get_historical_trades(
inst_id="BTC-USDT-SWAP",
start=start_time,
end=end_time
)
print(f"Fetched {len(trades)} trades")
print(trades.head() if not trades.empty else "No data")
# Fetch options chain
chain = client.get_options_chain_snapshot("BTC", "240329")
print(f"Options chain strikes: {len(chain.get('strikes', []))}")Performance Analysis and Visualization
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
def analyze_backtest_results(results_df: pd.DataFrame, trade_log_df: pd.DataFrame):
"""Generate comprehensive performance analysis."""
fig, axes = plt.subplots(3, 2, figsize=(16, 12))
fig.suptitle("OKX Volatility Arbitrage Strategy - Backtest Results", fontsize=14)
# 1. Equity Curve
ax1 = axes[0, 0]
ax1.plot(results_df["timestamp"], results_df["total_equity"],
label="Total Equity", color="blue", linewidth=1.5)
ax1.axhline(y=results_df["total_equity"].iloc[0], color="gray",
linestyle="--", alpha=0.5, label="Initial Capital")
ax1.set_title("Equity Curve")
ax1.set_ylabel("Portfolio Value (USD)")
ax1.legend()
ax1.grid(True, alpha=0.3)
# 2. Drawdown Chart
ax2 = axes[0, 1]
equity = results_df["total_equity"]
cummax = equity.cummax()
drawdown = (equity - cummax) / cummax * 100
ax2.fill_between(results_df["timestamp"], drawdown, 0,
color="red", alpha=0.3, label="Drawdown")
ax2.set_title(f"Drawdown (Max: {drawdown.min():.2f}%)")
ax2.set_ylabel("Drawdown (%)")
ax2.grid(True, alpha=0.3)
# 3. Returns Distribution
ax3 = axes[1, 0]
returns = results_df["return_pct"].diff().dropna() * 100
ax3.hist(returns, bins=50, color="steelblue", edgecolor="white", alpha=0.7)
ax3.axvline(x=returns.mean(), color="red", linestyle="--",
label=f"Mean: {returns.mean():.3f}%")
ax3.set_title("Returns Distribution")
ax3.set_xlabel("Return (%)")
ax3.set_ylabel("Frequency")
ax3.legend()
ax3.grid(True, alpha=0.3)
# 4. Trade Log Analysis
ax4 = axes[1, 1]
if not trade_log_df.empty:
trades_by_action = trade_log_df["action"].value_counts()
ax4.bar(trades_by_action.index, trades_by_action.values,
color=["green" if "BUY" in x else "red" for x in trades_by_action.index])
ax4.set_title("Trade Distribution by Action")
ax4.set_ylabel("Number of Trades")
else:
ax4.text(0.5, 0.5, "No trades executed", ha="center", va="center")
ax4.grid(True, alpha=0.3)
# 5. Spot Price vs PnL Correlation
ax5 = axes[2, 0]
ax5.scatter(results_df["spot_price"], results_df["market_pnl"],
alpha=0.5, s=10, c="purple")
ax5.set_title("Spot Price vs Market PnL")
ax5.set_xlabel("BTC Spot Price")
ax5.set_ylabel("Market PnL")
ax5.grid(True, alpha=0.3)
# 6. Rolling Sharpe Ratio
ax6 = axes[2, 1]
window = 20
rolling_returns = results_df["return_pct"].pct_change()
rolling_sharpe = (rolling_returns.rolling(window).mean() /
rolling_returns.rolling(window).std()) * np.sqrt(252)
ax6.plot(results_df["timestamp"], rolling_sharpe,
color="orange", linewidth=1.5, label=f"{window}-period Rolling Sharpe")
ax6.axhline(y=0, color="black", linestyle="-", alpha=0.3)
ax6.set_title("Rolling Sharpe Ratio")
ax6.set_ylabel("Sharpe Ratio")
ax6.legend()
ax6.grid(True, alpha=0.3