| Overall Statistics |
|
Total Orders 0 Average Win 0% Average Loss 0% Compounding Annual Return 0% Drawdown 0% Expectancy 0 Start Equity 1000000 End Equity 1000000 Net Profit 0% Sharpe Ratio 0 Sortino Ratio 0 Probabilistic Sharpe Ratio 0% Loss Rate 0% Win Rate 0% Profit-Loss Ratio 0 Alpha 0 Beta 0 Annual Standard Deviation 0 Annual Variance 0 Information Ratio -1.692 Tracking Error 0.141 Treynor Ratio 0 Total Fees $0.00 Estimated Strategy Capacity $0 Lowest Capacity Asset Portfolio Turnover 0% |
from AlgorithmImports import * class OilFuturesDeltaNeutralStrategy(QCAlgorithm): def Initialize(self): self.SetStartDate(2023, 1, 1) self.SetEndDate(2023, 4, 1) self.SetCash(1000000) # Add Crude Oil Futures self.future = self.AddFuture(Futures.Energies.CrudeOilWTI, Resolution.Minute) self.future.SetFilter(timedelta(80), timedelta(100)) # Slightly relaxed expiry filter # Add Future Options self.AddFutureOption(self.future.Symbol, self.OptionFilter) # Track price history for delta estimation self.prev_futures_prices = [] self.prev_option_prices = [] def OptionFilter(self, universe): return universe.Strikes(-10, 10) \ .Expiration(timedelta(80), timedelta(100)) \ .OnlyApplyFilterAtMarketOpen() def OnData(self, slice): if not slice.FutureChains: self.Log("No future chain data available.") return if not slice.OptionChains: self.Log("No option chain data available.") return # Step 1: Select 90-day Futures Contract future_chain = slice.FutureChains.get(self.future.Symbol) if not future_chain: self.Log("No valid futures contracts found.") return futures_contracts = sorted( [contract for contract in future_chain if 85 <= (contract.Expiry - self.Time).days <= 95], key=lambda x: abs((x.Expiry - self.Time).days - 90) ) if not futures_contracts: self.Log("No futures contract close to 90-day expiry found.") return selected_futures_contract = futures_contracts[0] futures_price = selected_futures_contract.LastPrice self.prev_futures_prices.append(futures_price) self.Log(f"Selected Futures Contract: {selected_futures_contract.Symbol} | Price: {futures_price}") # Step 2: Select ATM Option option_chain = slice.OptionChains.get(selected_futures_contract.Symbol) if not option_chain or len(option_chain) == 0: self.Log(f"No options found for futures contract {selected_futures_contract.Symbol}.") return atm_option_candidates = sorted( [contract for contract in option_chain if contract.OpenInterest > 10], key=lambda x: abs(x.Strike - futures_price) ) if not atm_option_candidates: self.Log("No ATM option found with sufficient liquidity.") return atm_option = atm_option_candidates[0] # Closest strike to futures price option_price = (atm_option.BidPrice + atm_option.AskPrice) / 2 self.prev_option_prices.append(option_price) self.Log(f"Selected ATM Option: {atm_option.Symbol} | Strike: {atm_option.Strike} | Price: {option_price}") # Ensure we have enough data for delta estimation if len(self.prev_futures_prices) > 5 and len(self.prev_option_prices) > 5: deltas = [ (self.prev_option_prices[i] - self.prev_option_prices[i - 1]) / (self.prev_futures_prices[i] - self.prev_futures_prices[i - 1]) for i in range(1, 5) if self.prev_futures_prices[i] != self.prev_futures_prices[i - 1] ] if deltas: estimated_delta = sum(deltas) / len(deltas) # Step 3: Compute Hedge Ratio if estimated_delta != 0: contracts_needed = round(-1 / estimated_delta) position_type = "Long" if estimated_delta < 0 else "Short" self.Log("\n====== Delta Hedge Calculation ======") self.Log(f"Date: {self.Time}") self.Log(f"Futures Price: {futures_price:.2f}") self.Log(f"Option Price: {option_price:.2f}") self.Log(f"Estimated Delta: {estimated_delta:.4f}") self.Log(f"Required Contracts: {contracts_needed}") self.Log(f"Suggested Position: {position_type} {abs(contracts_needed)} option contracts") # Keep only last 5 records for smoothing if len(self.prev_futures_prices) > 5: self.prev_futures_prices.pop(0) self.prev_option_prices.pop(0)
