From b76ee13cb4f8fc00482f6a521c0c066a1e48ad24 Mon Sep 17 00:00:00 2001 From: Brenden Matthews Date: Thu, 11 Jan 2024 08:08:20 -0500 Subject: [PATCH] Set call cap per symbol We can now set the call caps per symbol with `symbols..calls.cap_factor` and `symbols..calls.cap_target_floor`. Also fixed some typings. --- thetagang.toml | 11 ++++++++ thetagang/config.py | 2 ++ thetagang/portfolio_manager.py | 2 +- thetagang/util.py | 46 ++++++++++++++++++++++++++-------- 4 files changed, 49 insertions(+), 12 deletions(-) diff --git a/thetagang.toml b/thetagang.toml index 0c5f4c04e..77ba1d1e6 100644 --- a/thetagang.toml +++ b/thetagang.toml @@ -185,6 +185,9 @@ green = true # With covered calls, we can cap the number of calls to write by this factor. At # 1.0, we write covered calls on 100% of our positions. At 0.5, we'd only write # on 50% of our positions. This value must be between 1 and 0 inclusive. +# +# This can also be set per-symbol with +# `symbols..calls.cap_factor`. cap_factor = 1.0 # We may want to leave some percentage of our underlying stock perpetually @@ -197,6 +200,9 @@ cap_factor = 1.0 # and a smaller number (down to 0.0, 0%) is more bearish (i.e., cover all # positions with calls). This essentially sets a floor on the number of shares # we try to hold on to in order to avoid missing out on potential upside. +# +# This can also be set per-symbol with +# `symbols..calls.cap_target_floor`. cap_target_floor = 0.0 [write_when.puts] @@ -313,6 +319,11 @@ strike_limit = 100.0 # never write a call with a strike below $100 maintain_high_water_mark = true # maintain the high water mark when rolling calls +# These values can (optionally) be set on a per-symbol basis, in addition to +# `write_when.calls.cap_factor` and `write_when.calls.cap_target_floor. +cap_factor = 1.0 +cap_target_floor = 0.0 + [symbols.TLT] weight = 0.2 # parts = 20 diff --git a/thetagang/config.py b/thetagang/config.py index fbe224c6a..bbad37956 100644 --- a/thetagang/config.py +++ b/thetagang/config.py @@ -149,6 +149,8 @@ def validate_config(config): Optional("write_threshold_sigma"): And(float, lambda n: n > 0), Optional("strike_limit"): And(float, lambda n: n > 0), Optional("maintain_high_water_mark"): bool, + Optional("cap_factor"): And(float, lambda n: 0 <= n <= 1), + Optional("cap_target_floor"): And(float, lambda n: 0 <= n <= 1), }, Optional("puts"): { Optional("delta"): And(float, lambda n: 0 <= n <= 1), diff --git a/thetagang/portfolio_manager.py b/thetagang/portfolio_manager.py index 485397540..82ddd029a 100644 --- a/thetagang/portfolio_manager.py +++ b/thetagang/portfolio_manager.py @@ -764,7 +764,7 @@ def check_for_uncovered_positions(self, account_summary, portfolio_positions): ) target_short_calls = get_target_calls( - self.config, stock_count, self.target_quantities[symbol] + self.config, symbol, stock_count, self.target_quantities[symbol] ) new_contracts_needed = target_short_calls - short_call_count excess_calls = short_call_count - target_short_calls diff --git a/thetagang/util.py b/thetagang/util.py index a5ebf5a89..d42f2620a 100644 --- a/thetagang/util.py +++ b/thetagang/util.py @@ -2,7 +2,7 @@ from datetime import datetime from typing import Optional -from ib_insync import PortfolioItem, TagValue, util +from ib_insync import PortfolioItem, TagValue, Ticker, util from ib_insync.contract import Option from thetagang.options import option_dte @@ -101,31 +101,35 @@ def wait_n_seconds(pred, body, seconds_to_wait, started_at=None): wait_n_seconds(pred, body, seconds_to_wait, started_at) -def get_higher_price(ticker) -> float: +def get_higher_price(ticker: Ticker) -> float: # Returns the highest of either the option model price, the midpoint, or the # market price. The midpoint is usually a bit higher than the IB model's # pricing, but we want to avoid leaving money on the table in cases where # the spread might be messed up. This may in some cases make it harder for # orders to fill in a given day, but I think that's a reasonable tradeoff to # avoid leaving money on the table. - if ticker.modelGreeks: + if ticker.modelGreeks and ticker.modelGreeks.optPrice: return max([midpoint_or_market_price(ticker), ticker.modelGreeks.optPrice]) return midpoint_or_market_price(ticker) -def get_lower_price(ticker) -> float: +def get_lower_price(ticker: Ticker) -> float: # Same as get_highest_price(), except get the lower price instead. - if ticker.modelGreeks: + if ticker.modelGreeks and ticker.modelGreeks.optPrice: return min([midpoint_or_market_price(ticker), ticker.modelGreeks.optPrice]) return midpoint_or_market_price(ticker) -def midpoint_or_market_price(ticker) -> float: +def midpoint_or_market_price(ticker: Ticker) -> float: # As per the ib_insync docs, marketPrice returns the last price first, but # we often prefer the midpoint over the last price. This function pulls the # midpoint first, then falls back to marketPrice() if midpoint is nan. if util.isNan(ticker.midpoint()): - if util.isNan(ticker.marketPrice()) and ticker.modelGreeks: + if ( + util.isNan(ticker.marketPrice()) + and ticker.modelGreeks + and ticker.modelGreeks.optPrice + ): # Fallback to the model price if the greeks are available return ticker.modelGreeks.optPrice else: @@ -134,7 +138,7 @@ def midpoint_or_market_price(ticker) -> float: return ticker.midpoint() -def get_target_delta(config, symbol, right): +def get_target_delta(config: dict, symbol: str, right: str): p_or_c = "calls" if right.upper().startswith("C") else "puts" if ( p_or_c in config["symbols"][symbol] @@ -148,6 +152,24 @@ def get_target_delta(config, symbol, right): return config["target"]["delta"] +def get_cap_factor(config: dict, symbol: str): + if ( + "calls" in config["symbols"][symbol] + and "cap_factor" in config["symbols"][symbol]["calls"] + ): + return config["symbols"][symbol]["calls"]["cap_factor"] + return config["write_when"]["calls"]["cap_factor"] + + +def get_cap_target_floor(config: dict, symbol: str): + if ( + "calls" in config["symbols"][symbol] + and "cap_target_floor" in config["symbols"][symbol]["calls"] + ): + return config["symbols"][symbol]["calls"]["cap_target_floor"] + return config["write_when"]["calls"]["cap_target_floor"] + + def get_strike_limit(config: dict, symbol: str, right: str) -> Optional[float]: p_or_c = "calls" if right.upper().startswith("C") else "puts" if ( @@ -158,9 +180,11 @@ def get_strike_limit(config: dict, symbol: str, right: str) -> Optional[float]: return None -def get_target_calls(config: dict, current_shares: int, target_shares: int) -> int: - cap_factor = config["write_when"]["calls"]["cap_factor"] - cap_target_floor = config["write_when"]["calls"]["cap_target_floor"] +def get_target_calls( + config: dict, symbol: str, current_shares: int, target_shares: int +) -> int: + cap_factor = get_cap_factor(config, symbol) + cap_target_floor = get_cap_target_floor(config, symbol) min_uncovered = (target_shares * cap_target_floor) // 100 max_covered = (current_shares * cap_factor) // 100 total_coverable = current_shares // 100