github-featured/quantconnect--lean
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

ShortButterflyCall and ShortButterflyPut strategies helper factory methods (#7302)
Regression Tests / build (push) Has been cancelled
Benchmarks / build (push) Has been cancelled
Build & Test Lean / build (push) Has been cancelled
Research Regression Tests / build (push) Has been cancelled
Python Virtual Environments / build (push) Has been cancelled

Browse Source 
* Add ShortButterflyCall and ShortButterflyPut strategies helper factory methods

* Reduce duplication by adding the base OptionStrategyFactoryMethodsBaseAlgorithm algorithm class

* Housekeeping
This commit is contained in:
Jhonathan Abreu
2023-06-08 10:55:48 -04:00
committed by GitHub
parent cc9a061cb1
commit ad6046fea5
20 changed files with 1448 additions and 969 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Algorithm.Python/CoveredAndProtectiveCallStrategiesAlgorithm.py
+31 -62
View File
@@ -13,78 +13,47 @@
from AlgorithmImports import *
from OptionStrategyFactoryMethodsBaseAlgorithm import *
### <summary>
### This algorithm demonstrate how to use OptionStrategies helper class to batch send orders for common strategies.
### In this case, the algorithm tests the Covered and Protective Call strategies.
### </summary>
class CoveredAndProtectiveCallStrategiesAlgorithm(QCAlgorithm):
class CoveredAndProtectiveCallStrategiesAlgorithm(OptionStrategyFactoryMethodsBaseAlgorithm):
def Initialize(self):
self.SetStartDate(2015, 12, 24)
self.SetEndDate(2015, 12, 24)
self.SetCash(1000000)
def ExpectedOrdersCount(self) -> int:
return 4
option = self.AddOption("GOOG")
self._option_symbol = option.Symbol
def TradeStrategy(self, chain: OptionChain, option_symbol: Symbol):
contracts = sorted(sorted(chain, key = lambda x: abs(chain.Underlying.Price - x.Strike)),
key = lambda x: x.Expiry, reverse=True)
option.SetFilter(-2, +2, 0, 180)
if len(contracts) == 0: return
contract = contracts[0]
if contract != None:
self._covered_call = OptionStrategies.CoveredCall(option_symbol, contract.Strike, contract.Expiry)
self._protective_call = OptionStrategies.ProtectiveCall(option_symbol, contract.Strike, contract.Expiry)
self.Buy(self._covered_call, 2)
self.SetBenchmark("GOOG")
def AssertStrategyPositionGroup(self, positionGroup: IPositionGroup, option_symbol: Symbol):
positions = list(positionGroup.Positions)
if len(positions) != 2:
raise Exception(f"Expected position group to have 2 positions. Actual: {len(positions)}")
def OnData(self,slice):
if not self.Portfolio.Invested:
for kvp in slice.OptionChains:
chain = kvp.Value
contracts = sorted(sorted(chain, key = lambda x: abs(chain.Underlying.Price - x.Strike)),
key = lambda x: x.Expiry, reverse=True)
optionPosition = [position for position in positions if position.Symbol.SecurityType == SecurityType.Option][0]
if optionPosition.Symbol.ID.OptionRight != OptionRight.Call:
raise Exception(f"Expected option position to be a call. Actual: {optionPosition.Symbol.ID.OptionRight}")
if len(contracts) == 0: continue
contract = contracts[0]
if contract != None:
self._covered_call = OptionStrategies.CoveredCall(self._option_symbol, contract.Strike, contract.Expiry)
self._protective_call = OptionStrategies.ProtectiveCall(self._option_symbol, contract.Strike, contract.Expiry)
self.Buy(self._covered_call, 2)
else:
# Verify that the strategy was traded
positionGroup = list(self.Portfolio.Positions.Groups)[0]
underlyingPosition = [position for position in positions if position.Symbol.SecurityType == SecurityType.Equity][0]
expectedOptionPositionQuantity = -2
expectedUnderlyingPositionQuantity = 2 * self.Securities[option_symbol].SymbolProperties.ContractMultiplier
buyingPowerModel = positionGroup.BuyingPowerModel
if not isinstance(buyingPowerModel, OptionStrategyPositionGroupBuyingPowerModel):
raise Exception("Expected position group buying power model type: OptionStrategyPositionGroupBuyingPowerModel. "
f"Actual: {type(positionGroup.BuyingPowerModel).__name__}")
if optionPosition.Quantity != expectedOptionPositionQuantity:
raise Exception(f"Expected option position quantity to be {expectedOptionPositionQuantity}. Actual: {optionPosition.Quantity}")
positions = list(positionGroup.Positions)
if len(positions) != 2:
raise Exception(f"Expected position group to have 2 positions. Actual: {len(positions)}")
if underlyingPosition.Quantity != expectedUnderlyingPositionQuantity:
raise Exception(f"Expected underlying position quantity to be {expectedUnderlyingPositionQuantity}. Actual: {underlyingPosition.Quantity}")
optionPosition = [position for position in positions if position.Symbol.SecurityType == SecurityType.Option][0]
if optionPosition.Symbol.ID.OptionRight != OptionRight.Call:
raise Exception(f"Expected option position to be a call. Actual: {optionPosition.Symbol.ID.OptionRight}")
underlyingPosition = [position for position in positions if position.Symbol.SecurityType == SecurityType.Equity][0]
expectedOptionPositionQuantity = -2
expectedUnderlyingPositionQuantity = 2 * self.Securities[self._option_symbol].SymbolProperties.ContractMultiplier
if optionPosition.Quantity != expectedOptionPositionQuantity:
raise Exception(f"Expected option position quantity to be {expectedOptionPositionQuantity}. Actual: {optionPosition.Quantity}")
if underlyingPosition.Quantity != expectedUnderlyingPositionQuantity:
raise Exception(f"Expected underlying position quantity to be {expectedUnderlyingPositionQuantity}. Actual: {underlyingPosition.Quantity}")
# Now we should be able to close the position using the inverse strategy (a protective call)
self.Buy(self._protective_call, 2);
# We can quit now, no more testing required
self.Quit();
def OnEndOfAlgorithm(self):
if self.Portfolio.Invested:
raise Exception("Expected no holdings at end of algorithm")
orders_count = len(list(self.Transactions.GetOrders(lambda order: order.Status == OrderStatus.Filled)))
if orders_count != 4:
raise Exception("Expected 4 orders to have been submitted and filled, 2 for buying the covered call and 2 for the liquidation. "
f"Actual {orders_count}")
def OnOrderEvent(self, orderEvent):
self.Debug(str(orderEvent))
def LiquidateStrategy(self):
# We should be able to close the position using the inverse strategy (a protective call)
self.Buy(self._protective_call, 2)