Set call cap per symbol

We can now set the call caps per symbol with
`symbols.<symbol>.calls.cap_factor` and
`symbols.<symbol>.calls.cap_target_floor`.

Also fixed some typings.

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.<symbol>.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.<symbol>.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

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),

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

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