From 7c28aa305ebcab5ee78be6a0eca03377ece6437b Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 23 Jun 2026 05:07:51 +0800 Subject: [PATCH 1/2] Add live-gated default option overlay configs --- .../index_leaps_growth_overlay.zh-CN.md | 224 +++++ docs/us_equity_strategy_status.zh-CN.md | 5 +- src/us_equity_strategies/catalog.py | 19 +- .../entrypoints/__init__.py | 2 +- .../entrypoints/_common.py | 597 +----------- .../income_layer_defaults.py | 34 +- .../manifests/__init__.py | 19 +- src/us_equity_strategies/option_overlay.py | 884 ++++++++++++++++++ .../strategies/soxl_soxx_trend_income.py | 4 +- .../strategies/tqqq_growth_income.py | 14 +- tests/test_catalog.py | 84 +- tests/test_entrypoints.py | 74 +- tests/test_option_overlay.py | 290 ++++++ 13 files changed, 1575 insertions(+), 675 deletions(-) create mode 100644 docs/research/index_leaps_growth_overlay.zh-CN.md create mode 100644 src/us_equity_strategies/option_overlay.py create mode 100644 tests/test_option_overlay.py diff --git a/docs/research/index_leaps_growth_overlay.zh-CN.md b/docs/research/index_leaps_growth_overlay.zh-CN.md new file mode 100644 index 0000000..b81cc07 --- /dev/null +++ b/docs/research/index_leaps_growth_overlay.zh-CN.md @@ -0,0 +1,224 @@ +# 指数 LEAPS 增长增强层研究设计 + +_更新日期：2026-06-23_ + +> 投资有风险。本文是策略工程研究记录，不构成投资建议，也不代表任何账户应当交易期权。 + +## 初步结论 + +买入 `QQQ` / `SPY` LEAPS call 可以作为“增长增强层”研究，但不应直接替代当前收入层。 + +当前收入层的目标是降低账户级压力回撤：`SCHD`、`DGRO`、`SGOV`、`SPYI`、`QQQI` 等资产按 `log_total_drawdown_budget` 反推比例，服务于大账户波动钝化。LEAPS call 的收益来源不同，它是用有限权利金换取长期指数上涨凸性，最大亏损通常是权利金，组合效果更像受限预算的再杠杆，而不是稳定收入。 + +因此建议把结构拆成两层： + +| 层 | 目标 | 默认处理 | +| --- | --- | --- | +| 收入 / 稳定层 | 降低总组合压力回撤，保留现金流和防守资产 | 保持现有 `income_layer_*` 默认，不把 LEAPS 放入 `income_layer_allocations` | +| LEAPS 增长增强层 | 用小额权利金预算增加 `QQQ` / `SPY` 长期上涨凸性 | 默认配置可见且总开关开启；配方未晋级前由 live gate 阻断，不产生真实订单意图 | + +## 外部事实核对 + +- Cboe 对 LEAPS 的定位是长期期权，最长可到约 3 年，买方以权利金换取长期方向性参与权，而不是持有股票本身。 +- OCC 风险披露口径下，买入期权的最大损失一般是已支付权利金和交易成本；但权利金可以全部损失，期权也可能到期归零。 +- `SPY` 和 `QQQ` 都是高流动性宽基 ETF，适合作为期权研究底层资产；但期权回测必须使用合约链、bid/ask、希腊值、到期日和分红 / 利率假设，不能只用 ETF 收盘价替代。 +- 公开研究里也有反例：长期深度价内 call 并不天然优于股票，尤其在下跌和高隐含波动环境中，时间价值损耗与入场估值会明显影响结果。 + +参考资料： + +- https://www.cboe.com/tradable_products/equity_indices/leaps_options/specifications/ +- https://www.theocc.com/company-information/documents-and-archives/options-disclosure-document +- https://www.ssga.com/us/en/intermediary/etfs/resources/doc-viewer#spy&prospectus +- https://www.invesco.com/qqq-etf/en/home.html +- https://papers.ssrn.com/sol3/papers.cfm?abstract_id=1443511 + +## 仓库现状 + +`UsEquityStrategies` 已经有独立的 option overlay 执行意图框架： + +- `tqqq_leaps_growth_v1`、`qqq_leaps_growth_v1`、`spy_leaps_growth_v1` 和 + `soxx_put_credit_spread_income_v1` 都只作为 `option_overlay.py` 内的 research candidate。 +- 组合型 live profile 已携带默认启用的 `option_*` 配置；但这些 recipe 仍是 research candidate， + `promotion_evidence = false`，entrypoint 会给出 `research_only_recipe` 跳过原因，不生成 + `option_order_intents`。 +- `option_overlay.py` 会按 option chain 生成 `option_order_intents`，包括 LEAPS call、put credit spread 的筛选、预算和跳过原因；`entrypoints/_common.py` 只保留兼容导出和配置弹出工具。 + +本次补充： + +- 新增 `spy_leaps_growth_v1`，允许 runtime config 选择 `SPY` LEAPS call。 +- 将 option overlay recipe、合约链解析和意图生成从 entrypoint common 工具拆出到独立模块，保持收入层和期权层职责分离。 +- 默认 profile 携带期权 overlay 配置，便于设置面和诊断统一；但 live gate 在真实期权链 evidence + 通过前保持关闭，避免未经验证改变实盘行为。 + +## 候选设计 + +### `index_leaps_light` + +适合先做 paper / advisory 观察。 + +| 参数 | 建议 | +| --- | --- | +| 底层 | `QQQ` 或 `SPY`，不优先使用 `TQQQ` | +| 权利金预算 | 每个 underlier `1%`-`2%` NAV，组合总预算不超过 `3%` NAV | +| 到期 | 目标 `24` 个月，接受 `540`-`930` DTE | +| Delta | 目标 `0.70`-`0.80`，默认 `0.75` | +| 入场 | 底层在 200 日均线上方，63 日动量为正；或者大跌后重新收复长期趋势线 | +| 滚动 | DTE 低于 `12` 个月时滚动 | +| 止盈 | 合约价值达到成本约 `2x` 且数量允许时，卖出足够合约回收本金，保留 runner | +| 流动性 | bid/ask spread 占 mid 不超过 `8%`-`12%` | + +### `index_leaps_balanced` + +适合验证后给增长型账户使用。 + +| 参数 | 建议 | +| --- | --- | +| 底层 | `QQQ` 为主，`SPY` 分散；实现上可先用单一 recipe，后续再支持多底层篮子 | +| 权利金预算 | 总预算 `3%` NAV，硬上限 `5%` NAV | +| 入场 | 必须满足趋势门控；不在 MA200 下方摊低 LEAPS | +| 与收入层关系 | 保留收入层，不从 `SGOV` 防守资产里强行挪钱；权利金预算应来自增长预算 | + +### `crash_reentry_leaps` + +适合研究“熊市后重新站上趋势线”的高赔率窗口。 + +| 参数 | 建议 | +| --- | --- | +| 触发 | `QQQ` 或 `SPY` 从 `20%+` 回撤后重新站上 MA200，且 63 日动量转正 | +| 预算 | 首次 `1%` NAV，确认后最多加到 `3%` NAV | +| 目的 | 参与大级别修复，而不是日常持有 LEAPS | + +## 回测验证口径 + +正式 promotion evidence 应放在 `UsEquitySnapshotPipelines`，而不是只在策略仓库写结论。 + +### 输入数据 + +必须有真实期权链历史，至少包括： + +- 底层 ETF 复权价格：`SPY`、`QQQ`。 +- 合约级日线：expiration、strike、right、bid、ask、mid/mark、volume、open interest。 +- 希腊值：delta、theta、vega；没有 vendor greeks 时要固定模型重算并记录参数。 +- 利率与分红假设：短债利率、ETF 分红收益率。 +- 交易假设：开仓用 ask 或 mid+滑点，平仓用 bid 或 mid-滑点，合约乘数 100，佣金可单独建列。 + +### 代理验证 + +如果暂时拿不到历史期权链，可以先做 Black-Scholes 代理，但只能用于筛方向，不可作为 promotion evidence： + +- 用 `SPY` / `QQQ` 复权价格生成 252 日 realized vol。 +- 用 `realized_vol * IV multiplier` 估计隐含波动，设置 floor/cap。 +- 每次入场反解目标 delta strike。 +- 每日按剩余 DTE、底层价格、估计 IV 重新定价。 +- 明确记录误差：忽略真实 skew、合约流动性、早期行权、分红变化和实盘 bid/ask。 + +### 对照组 + +至少比较这些组： + +- 当前默认收入层，无 LEAPS。 +- 当前默认收入层 + `QQQ` LEAPS 1% / 3% / 5% NAV budget。 +- 当前默认收入层 + `SPY` LEAPS 1% / 3% / 5% NAV budget。 +- 只买 `QQQ` / `SPY`。 +- 当前 `tqqq_growth_income` / `russell_top50_leader_rotation` 默认策略。 +- 现有 `soxx_put_credit_spread_income_v1`，作为“卖方收入”对照。 + +### 指标 + +推广前必须同时看收益和失败路径： + +- CAGR、总收益、最大回撤、Calmar。 +- 最差 1 年 / 3 年滚动收益。 +- 相对 `QQQ` / `SPY` 的超额 CAGR 和回撤差。 +- 权利金归零次数、连续亏损开仓次数、平均持有期、到期前滚动比例。 +- 最差单笔 LEAPS 亏损、最佳单笔收益、2x 回本触发率。 +- 年化 turnover、bid/ask 成本、无法成交 / 合约不满足流动性筛选的比例。 +- 与收入层相互影响：收入层比例、现金 / SGOV 比例、增长核心仓位被挤占程度。 + +## 风控边界 + +- LEAPS 默认不属于 `income_layer_allocations`。 +- live 默认配置可以携带 `option_*` key；只有 `OPTION_OVERLAY_RESEARCH_CANDIDATES` 中对应 recipe + 晋级为 `status = live` 且 `promotion_evidence = true` 后才允许生成真实订单意图。 +- 未有真实期权链证据前，LEAPS overlay 只能是 `shadow` / `paper` / `advisory`。 +- 默认权利金预算不超过 `3%` NAV；研究可扫到 `5%`，但超过 `5%` 应视为显著再杠杆。 +- 优先研究 `QQQ` / `SPY`，不优先推广 `TQQQ` LEAPS。`TQQQ` 本身已是杠杆 ETF，LEAPS 叠加后路径风险、波动率定价和流动性风险都更复杂。 +- 不做裸卖期权；卖方收入策略必须是定义风险结构，例如 put credit spread。 +- 任何默认切换必须附带短、中、长窗口 evidence，并通过同仓库的 contract / entrypoint 测试。 + +## Runtime 配置面 + +期权层现在有一个总开关和两个分层开关： + +| 配置 | 作用 | +| --- | --- | +| `option_overlay_enabled` | 总开关；设为 `false` 时关闭所有期权 overlay，不生成 `option_order_intents`。 | +| `option_growth_overlay_enabled` | 增长增强层开关，例如 `QQQ` / `SPY` LEAPS call。 | +| `option_growth_overlay_recipe` | 增长增强层 recipe，例如 `qqq_leaps_growth_v1`、`spy_leaps_growth_v1`。 | +| `option_growth_overlay_start_usd` | 账户权益达到该阈值后才激活增长增强层。 | +| `option_growth_overlay_nav_budget_ratio` | LEAPS 权利金预算占 NAV 比例；实现侧硬限制不超过 `10%`。 | +| `option_income_overlay_enabled` | 期权收入层开关，例如定义风险 put credit spread。 | +| `option_income_overlay_recipe` | 期权收入层 recipe，例如 `soxx_put_credit_spread_income_v1`。 | +| `option_income_overlay_start_usd` | 账户权益达到该阈值后才激活期权收入层。 | +| `option_income_overlay_nav_risk_ratio` | 卖方结构最大风险预算占 NAV 比例；实现侧硬限制不超过 `2%`。 | + +live 默认：组合型 profile 会带默认启用的期权层设置，但当前所有 recipe 仍是 research candidate。 +因此 entrypoint 会生成期权层诊断，不会生成真实 `option_order_intents`；执行效果等同关闭，但设置面能提前固定。 + +| Profile | 默认期权层 | 默认 recipe | 起点 | 预算 | +| --- | --- | --- | ---: | ---: | +| `global_etf_rotation` | growth overlay | `spy_leaps_growth_v1` | `500000` | `1.5%` NAV premium | +| `tqqq_growth_income` | growth overlay | `tqqq_leaps_growth_v1` | `250000` | `3%` NAV premium | +| `soxl_soxx_trend_income` | income overlay | `soxx_put_credit_spread_income_v1` | `150000` | `1%` NAV max loss | +| `russell_top50_leader_rotation` | growth overlay | `spy_leaps_growth_v1` | `300000` | `1.5%` NAV premium | + +`nasdaq_sp500_smart_dca` / `ibit_smart_dca` 是只买不卖的定投 profile，默认不带直接期权 overlay。 + +显式关闭期权层的兼容写法： + +```json +{ + "option_overlay_enabled": false +} +``` + +示例 1：打开较保守的 `SPY` LEAPS 增长增强层，把权利金预算压到 `1.5%` NAV： + +```json +{ + "option_overlay_enabled": true, + "option_growth_overlay_enabled": true, + "option_growth_overlay_recipe": "spy_leaps_growth_v1", + "option_growth_overlay_start_usd": 250000, + "option_growth_overlay_nav_budget_ratio": 0.015 +} +``` + +示例 2：打开成长型 `QQQ` LEAPS 增长增强层，权利金预算为 `3%` NAV： + +```json +{ + "option_overlay_enabled": true, + "option_growth_overlay_enabled": true, + "option_growth_overlay_recipe": "qqq_leaps_growth_v1", + "option_growth_overlay_start_usd": 250000, + "option_growth_overlay_nav_budget_ratio": 0.03 +} +``` + +LEAPS 开仓不是盲买 call。`SPY` / `QQQ` recipe 会先看 entrypoint 的 `signal` / `regime`， +并在底层指标存在时要求长期趋势和中期动量通过，例如 `above_200dma` / `sma200_pass` +为真且 `momentum_63d` 为正；否则只记录 skip，不生成买入意图。 + +## 建议路线 + +1. 保留现有收入层默认。 +2. 用 `spy_leaps_growth_v1` / `qqq_leaps_growth_v1` 做 runtime override 或 paper signal 观察。 +3. 使用 `UsEquitySnapshotPipelines` 的 `useq-research-index-leaps-growth-overlay` 先跑 + Black-Scholes proxy research；该输出只用于方向筛选，不能作为 promotion evidence。拿到真实 + 历史期权链后，用同一 CLI 的 `--mode option-chain --option-chain ` 跑 bid/ask 链路， + 作为后续 promotion review 的证据输入。 +4. 若 `QQQ` / `SPY` LEAPS 在 2000、2008、2020、2022 等窗口的 worst-case 可接受，再把对应 + recipe 晋级为 `status = live` 且写入 `promotion_evidence = true`；优先从宽基 + `qqq_leaps_growth_v1` / `spy_leaps_growth_v1` 开始，不优先启用 `TQQQ` LEAPS。 +5. 若结果只是在牛市提高 CAGR、但最差 1-3 年窗口明显恶化，则保持为手动 overlay，不进入默认配置。 diff --git a/docs/us_equity_strategy_status.zh-CN.md b/docs/us_equity_strategy_status.zh-CN.md index 9de02b1..02b0eea 100644 --- a/docs/us_equity_strategy_status.zh-CN.md +++ b/docs/us_equity_strategy_status.zh-CN.md @@ -1,6 +1,6 @@ # 美股策略状态与研究手册 -_更新日期：2026-06-19_ +_更新日期：2026-06-23_ 这份文档只记录当前可配置的美股策略 profile、输入形态和研究状态，不记录任何账户或服务正在运行的 profile。部署单元当前跑什么属于私有运行信息，应留在云端配置或私有运行记录里。 @@ -39,6 +39,8 @@ _更新日期：2026-06-19_ 除 `nasdaq_sp500_smart_dca` / `ibit_smart_dca` 这类只买不卖的现金定投 profile 外，保留的组合型 runtime profile 默认都启用收入层，且下游策略配置可以覆盖任意 `income_layer_*` 参数；需要关闭时设置 `income_layer_enabled = false`。当前默认统一使用 `log_total_drawdown_budget`，先按账户规模给出目标总回撤预算，再用核心策略压力回撤和收入篮子压力回撤反推出收入层比例。`income_layer_activation_band_ratio` 会在 `start` 到 `start * (1 + band)` 之间把正常目标比例从 0 平滑放大到 1，避免门槛附近来回卡住。 +这些默认收入层是 live 配置：它们来自已归档的收入层回测 / 复核证据，并继续作为普通 ETF 目标仓位执行。`SPYI` / `QQQI` 留在收入层，因为策略只买卖 ETF 本身，不直接选择期权合约。直接期权层现在有默认设置，但所有未通过真实期权链验证的 recipe 都保持 research-only；配置默认可见，总开关默认开启，live gate 会在 `promotion_evidence = false` 时阻断真实期权订单意图。 + | Profile | 模式 | 起点 | 平滑带 | 硬上限 | 默认收入篮子 | | --- | --- | ---: | ---: | ---: | --- | | `tqqq_growth_income` | `log_total_drawdown_budget` | `250000` | `20%` | `55%` | `SCHD 30% / DGRO 20% / SGOV 40% / SPYI 8% / QQQI 2%` | @@ -71,6 +73,7 @@ _更新日期：2026-06-19_ | --- | --- | --- | | `crisis_response_shadow` 插件 | 可作为 `tqqq_growth_income` 的 `shadow` 插件候选，只写信号、日志和通知上下文。 | 现在是 defense-only 黑天鹅观察流，不下单、不改 allocation；需要稳定 shadow 日志后再做 evidence review。 | | 事件反弹 / MAGS 路线 | 保持 research-only，不作为运行策略 profile。 | 对 MAGS 的正贡献不稳定，且事件反弹预算不应该混进黑天鹅逃命插件。 | +| `QQQ` / `SPY` LEAPS 增长增强层 | 已有 option overlay 意图框架；组合型 live profile 默认带 `option_*` 设置，但 `spy_leaps_growth_v1` / `qqq_leaps_growth_v1` 等 recipe 仍是 research candidate，当前会以 `research_only_recipe` 跳过，不产生真实订单意图，研究设计见 [`docs/research/index_leaps_growth_overlay.zh-CN.md`](./research/index_leaps_growth_overlay.zh-CN.md)。 | 属于有限权利金预算的增长增强层，不是当前低回撤收入层的直接替代；需要真实期权链回测后才能把 recipe 晋级为 live。 | | AI 审计 / AI 上下文 | 不进入交易路径。 | 回测结果来自确定性指标，不依赖 AI；AI 可以辅助离线 review、总结新闻或检查文档，但不能作为自动买卖开关。 | | Russell 1000 代理长周期回测 | 研究待补。 | 2017 年前缺少可靠 point-in-time Russell 1000 / Top50 数据，需要代理构造并明确后视偏差。 | | Russell Top50 leader rotation paper 观察 | 当前保留候选。 | 历史结果强，且无杠杆，但 Top2 袖子仍有集中风险；要确认 snapshot、整数股、换手和通知稳定。 | diff --git a/src/us_equity_strategies/catalog.py b/src/us_equity_strategies/catalog.py index 77ec66b..247fb1b 100644 --- a/src/us_equity_strategies/catalog.py +++ b/src/us_equity_strategies/catalog.py @@ -20,6 +20,7 @@ from .ai_extensions import build_default_ai_extension_config from .income_layer_defaults import income_layer_default_config +from .option_overlay import option_overlay_default_config GLOBAL_ETF_ROTATION_PROFILE = "global_etf_rotation" # Legacy alias retained for lookups and docs; runtime registry is canonical rotation. @@ -106,6 +107,7 @@ "confidence_volatility_window": 126, "confidence_volatility_max_ratio": 1.3, **income_layer_default_config(GLOBAL_ETF_ROTATION_PROFILE), + **option_overlay_default_config(GLOBAL_ETF_ROTATION_PROFILE), "market_regime_control_enabled": True, "market_regime_control_apply_risk_reduced": False, "market_regime_control_apply_risk_off": False, @@ -115,16 +117,11 @@ TQQQ_GROWTH_INCOME_PROFILE: { "benchmark_symbol": "QQQ", "managed_symbols": ("TQQQ", "QQQM", "BOXX", "SCHD", "DGRO", "SGOV", "SPYI", "QQQI"), - "income_threshold_usd": 250000.0, - "qqqi_income_ratio": 0.10, "cash_reserve_ratio": 0.02, "rebalance_threshold_ratio": 0.01, "execution_cash_reserve_ratio": 0.0, **income_layer_default_config(TQQQ_GROWTH_INCOME_PROFILE), - "option_growth_overlay_enabled": True, - "option_growth_overlay_recipe": "tqqq_leaps_growth_v1", - "option_growth_overlay_start_usd": 250000.0, - "option_growth_overlay_nav_budget_ratio": 0.03, + **option_overlay_default_config(TQQQ_GROWTH_INCOME_PROFILE), "attack_allocation_mode": "fixed_qqq_tqqq_pullback", "dual_drive_qqq_weight": 0.45, "dual_drive_tqqq_weight": 0.45, @@ -165,10 +162,7 @@ "min_trade_floor": 100.0, "rebalance_threshold_ratio": 0.01, **income_layer_default_config(SOXL_SOXX_TREND_INCOME_PROFILE), - "option_income_overlay_enabled": True, - "option_income_overlay_recipe": "soxx_put_credit_spread_income_v1", - "option_income_overlay_start_usd": 1000000.0, - "option_income_overlay_nav_risk_ratio": 0.01, + **option_overlay_default_config(SOXL_SOXX_TREND_INCOME_PROFILE), "trend_entry_buffer": 0.08, "trend_mid_buffer": 0.06, "trend_exit_buffer": 0.02, @@ -226,10 +220,7 @@ "runtime_execution_window_trading_days": 3, "execution_cash_reserve_ratio": 0.0, **income_layer_default_config(RUSSELL_TOP50_LEADER_ROTATION_PROFILE), - "option_growth_overlay_enabled": True, - "option_growth_overlay_recipe": "qqq_leaps_growth_v1", - "option_growth_overlay_start_usd": 1000000.0, - "option_growth_overlay_nav_budget_ratio": 0.03, + **option_overlay_default_config(RUSSELL_TOP50_LEADER_ROTATION_PROFILE), "market_regime_control_enabled": True, "market_regime_control_apply_risk_reduced": False, "market_regime_control_apply_risk_off": False, diff --git a/src/us_equity_strategies/entrypoints/__init__.py b/src/us_equity_strategies/entrypoints/__init__.py index e2c72d8..c1e09b7 100644 --- a/src/us_equity_strategies/entrypoints/__init__.py +++ b/src/us_equity_strategies/entrypoints/__init__.py @@ -25,6 +25,7 @@ soxl_soxx_trend_income_manifest, tqqq_growth_income_manifest, ) +from us_equity_strategies.option_overlay import build_option_overlay_diagnostics from us_equity_strategies.strategies import ( global_etf_rotation as legacy_global_etf_rotation, ibit_smart_dca as ibit_smart_dca_strategy, @@ -39,7 +40,6 @@ apply_reserved_cash_policy_to_usd_config, apply_income_layer_to_weights, apply_market_regime_control_to_weights, - build_option_overlay_diagnostics, default_signal_text_fn, default_translator, get_current_holdings, diff --git a/src/us_equity_strategies/entrypoints/_common.py b/src/us_equity_strategies/entrypoints/_common.py index e6396be..00503ef 100644 --- a/src/us_equity_strategies/entrypoints/_common.py +++ b/src/us_equity_strategies/entrypoints/_common.py @@ -1,7 +1,6 @@ from __future__ import annotations from collections.abc import Mapping -from datetime import date, datetime from quant_platform_kit.strategy_contracts import PositionTarget, StrategyContext from us_equity_strategies.income_layer import ( @@ -11,6 +10,11 @@ from us_equity_strategies.market_regime_control_contract import ( resolve_market_regime_position_control_authorization, ) +from us_equity_strategies.option_overlay import ( + OPTION_OVERLAY_CONFIG_KEYS as OPTION_OVERLAY_CONFIG_KEYS, + OPTION_OVERLAY_RECIPE_DETAILS as OPTION_OVERLAY_RECIPE_DETAILS, + build_option_overlay_diagnostics as build_option_overlay_diagnostics, +) SAFE_HAVENS = {"BIL", "BOXX"} @@ -44,61 +48,8 @@ "market_regime_control_risk_off_scalar", "market_regime_control_safe_haven", } -OPTION_OVERLAY_CONFIG_KEYS = { - "option_growth_overlay_enabled", - "option_growth_overlay_recipe", - "option_growth_overlay_start_usd", - "option_growth_overlay_nav_budget_ratio", - "option_income_overlay_enabled", - "option_income_overlay_recipe", - "option_income_overlay_start_usd", - "option_income_overlay_nav_risk_ratio", -} MARKET_REGIME_CONTROL_PROFILE = "market_regime_control" MARKET_REGIME_POSITION_ROUTES = frozenset({"risk_reduced", "risk_off"}) -OPTION_OVERLAY_RECIPE_DETAILS = { - "tqqq_leaps_growth_v1": { - "structure": "long_call_leaps", - "underlier": "TQQQ", - "premium_budget_ratio": 0.03, - "target_delta": 0.75, - "target_dte_months": 24, - "roll_dte_months": 12, - "contract_multiplier": 100, - "min_dte_days": 540, - "max_dte_days": 930, - "max_bid_ask_spread_ratio": 0.12, - "entry_gate": "core_risk_on_and_qqq_above_200dma_and_qqq_63d_momentum_positive", - "principal_take_profit": "sell_enough_contracts_to_recover_cost_after_2x_when_contract_count_allows", - }, - "qqq_leaps_growth_v1": { - "structure": "long_call_leaps", - "underlier": "QQQ", - "premium_budget_ratio": 0.03, - "target_delta": 0.75, - "target_dte_months": 24, - "roll_dte_months": 12, - "contract_multiplier": 100, - "min_dte_days": 540, - "max_dte_days": 930, - "max_bid_ask_spread_ratio": 0.12, - "entry_gate": "qqq_above_200dma_and_qqq_63d_momentum_positive", - "principal_take_profit": "sell_enough_contracts_to_recover_cost_after_2x_when_contract_count_allows", - }, - "soxx_put_credit_spread_income_v1": { - "structure": "put_credit_spread", - "underlier": "SOXX", - "target_dte_days": 45, - "short_put_otm_pct": 0.08, - "long_put_otm_pct": 0.18, - "max_loss_budget_ratio": 0.01, - "contract_multiplier": 100, - "min_dte_days": 25, - "max_dte_days": 65, - "max_iv_rank": 0.80, - "entry_gate": "soxx_trend_positive_and_iv_rank_not_extreme", - }, -} def merge_runtime_config(manifest_default_config: Mapping[str, object], ctx: StrategyContext) -> dict[str, object]: @@ -337,544 +288,6 @@ def resolve_market_regime_control_context(ctx: StrategyContext) -> dict[str, obj ) -def _effective_option_recipe_detail( - family: str, - recipe_detail: Mapping[str, object], - option_overlay_config: Mapping[str, object], -) -> dict[str, object]: - effective = dict(recipe_detail) - if family == "growth": - default = _as_float(recipe_detail.get("premium_budget_ratio"), default=0.03) - effective["premium_budget_ratio"] = _clamped_ratio( - option_overlay_config.get("option_growth_overlay_nav_budget_ratio"), - default=default, - upper=0.10, - ) - elif family == "income": - default = _as_float(recipe_detail.get("max_loss_budget_ratio"), default=0.01) - effective["max_loss_budget_ratio"] = _clamped_ratio( - option_overlay_config.get("option_income_overlay_nav_risk_ratio"), - default=default, - upper=0.02, - ) - return effective - - -def _parse_date(value: object) -> date | None: - if value is None or value == "": - return None - if isinstance(value, datetime): - return value.date() - if isinstance(value, date): - return value - text = str(value).strip() - if not text: - return None - if len(text) == 8 and text.isdigit(): - text = f"{text[:4]}-{text[4:6]}-{text[6:]}" - try: - return datetime.fromisoformat(text[:10]).date() - except ValueError: - return None - - -def _as_of_date(ctx: StrategyContext) -> date: - parsed = _parse_date(ctx.as_of) - return parsed or datetime.utcnow().date() - - -def _normalize_right(value: object) -> str: - text = str(value or "").strip().upper() - if text in {"CALL", "C"}: - return "C" - if text in {"PUT", "P"}: - return "P" - return text - - -def _option_chain_payload(ctx: StrategyContext, underlier: str) -> object: - underlier = str(underlier or "").strip().upper() - for source in (ctx.market_data, ctx.runtime_config): - payload = source.get("option_chains") if isinstance(source, Mapping) else None - if isinstance(payload, Mapping): - direct = payload.get(underlier) or payload.get(underlier.lower()) - if direct is not None: - return direct - payload = source.get("option_chain") if isinstance(source, Mapping) else None - if isinstance(payload, Mapping): - payload_underlier = str(payload.get("underlier") or payload.get("symbol") or "").strip().upper() - if payload_underlier == underlier: - return payload - elif payload: - return payload - return None - - -def _chain_rows(payload: object) -> tuple[Mapping[str, object], ...]: - if payload is None: - return () - if isinstance(payload, Mapping): - for key in ("contracts", "options", "rows", "chain"): - rows = payload.get(key) - if rows is not None: - return tuple(row for row in rows if isinstance(row, Mapping)) - return () - try: - return tuple(row for row in payload if isinstance(row, Mapping)) # type: ignore[union-attr] - except TypeError: - return () - - -def _chain_spot(payload: object, underlier: str, ctx: StrategyContext) -> float: - if isinstance(payload, Mapping): - for key in ("spot", "underlier_price", "underlying_price", "last_price"): - price = _as_float(payload.get(key), default=0.0) - if price > 0.0: - return price - quote = ctx.market_data.get(str(underlier).strip().upper()) - if isinstance(quote, Mapping): - for key in ("last_price", "last", "close", "price"): - price = _as_float(quote.get(key), default=0.0) - if price > 0.0: - return price - return 0.0 - - -def _row_date(row: Mapping[str, object]) -> date | None: - for key in ("expiration", "expiry", "lastTradeDateOrContractMonth", "last_trade_date"): - parsed = _parse_date(row.get(key)) - if parsed is not None: - return parsed - return None - - -def _row_delta(row: Mapping[str, object]) -> float | None: - for key in ("delta", "model_delta"): - if key in row: - return _as_float(row.get(key), default=0.0) - greeks = row.get("greeks") - if isinstance(greeks, Mapping) and "delta" in greeks: - return _as_float(greeks.get("delta"), default=0.0) - return None - - -def _row_bid_ask_mid(row: Mapping[str, object]) -> tuple[float, float, float]: - bid = _as_float(row.get("bid") or row.get("bid_price"), default=0.0) - ask = _as_float(row.get("ask") or row.get("ask_price"), default=0.0) - raw_mid = _as_float(row.get("mid") or row.get("mark") or row.get("last"), default=0.0) - if bid > 0.0 and ask > 0.0: - return bid, ask, (bid + ask) / 2.0 - if raw_mid > 0.0 and ask <= 0.0: - ask = raw_mid - if raw_mid > 0.0: - return bid, ask, raw_mid - return bid, ask, ask if ask > 0.0 else bid - - -def _normalize_option_positions(ctx: StrategyContext, underlier: str, right: str) -> tuple[Mapping[str, object], ...]: - portfolio = ctx.portfolio - metadata = getattr(portfolio, "metadata", {}) if portfolio is not None else {} - rows = metadata.get("option_positions") if isinstance(metadata, Mapping) else () - normalized = [] - for row in rows or (): - if not isinstance(row, Mapping): - continue - row_underlier = str(row.get("underlier") or row.get("symbol") or "").strip().upper() - row_right = _normalize_right(row.get("right") or row.get("put_call")) - quantity = _as_float(row.get("quantity"), default=0.0) - if row_underlier == underlier and row_right == right and quantity > 0.0: - normalized.append(row) - return tuple(normalized) - - -def _long_call_entry_gate_passed(recipe: str, base_diagnostics: Mapping[str, object]) -> tuple[bool, str | None]: - signal_context = base_diagnostics.get("notification_context") - if isinstance(signal_context, Mapping): - raw_signal = signal_context.get("signal") - if isinstance(raw_signal, Mapping): - state = str(raw_signal.get("state") or "").strip().lower() - if state: - return state in {"entry", "hold"}, None if state in {"entry", "hold"} else "entry_gate_not_met" - regime = str(base_diagnostics.get("regime") or "").strip().lower() - if recipe == "qqq_leaps_growth_v1" and regime: - return regime == "risk_on", None if regime == "risk_on" else "entry_gate_not_met" - return True, None - - -def _put_credit_spread_gate_passed( - payload: object, - recipe_detail: Mapping[str, object], - base_diagnostics: Mapping[str, object], -) -> tuple[bool, str | None]: - blend_tier = str(base_diagnostics.get("blend_tier") or "").strip().lower() - if blend_tier and blend_tier not in {"full", "mid"}: - return False, "entry_gate_not_met" - iv_rank = None - if isinstance(payload, Mapping): - for key in ("iv_rank", "implied_volatility_rank"): - if key in payload: - iv_rank = _as_float(payload.get(key), default=-1.0) - break - if iv_rank is not None and iv_rank > _as_float(recipe_detail.get("max_iv_rank"), default=0.80): - return False, "iv_rank_too_high" - return True, None - - -def _append_skip(intents_payload: dict[str, object], *, recipe: str, reason: str, underlier: str) -> None: - skipped = list(intents_payload.get("skipped") or ()) - skipped.append({"recipe": recipe, "underlier": underlier, "reason": reason}) - intents_payload["skipped"] = skipped - - -def _build_long_call_intents( - *, - recipe: str, - recipe_detail: Mapping[str, object], - ctx: StrategyContext, - total_equity: float, - base_diagnostics: Mapping[str, object], - intents_payload: dict[str, object], -) -> None: - underlier = str(recipe_detail.get("underlier") or "").strip().upper() - gate_passed, gate_reason = _long_call_entry_gate_passed(recipe, base_diagnostics) - if not gate_passed: - _append_skip(intents_payload, recipe=recipe, underlier=underlier, reason=gate_reason or "entry_gate_not_met") - return - - payload = _option_chain_payload(ctx, underlier) - rows = _chain_rows(payload) - if not rows: - _append_skip(intents_payload, recipe=recipe, underlier=underlier, reason="missing_option_chain") - return - - as_of = _as_of_date(ctx) - target_dte = int(_as_float(recipe_detail.get("target_dte_months"), default=24.0) * 30.4375) - target_delta = _as_float(recipe_detail.get("target_delta"), default=0.75) - min_dte = int(_as_float(recipe_detail.get("min_dte_days"), default=540.0)) - max_dte = int(_as_float(recipe_detail.get("max_dte_days"), default=930.0)) - max_spread_ratio = _as_float(recipe_detail.get("max_bid_ask_spread_ratio"), default=0.12) - multiplier = int(_as_float(recipe_detail.get("contract_multiplier"), default=100.0)) - candidates = [] - for row in rows: - if _normalize_right(row.get("right") or row.get("type")) != "C": - continue - expiration = _row_date(row) - if expiration is None: - continue - dte = (expiration - as_of).days - if dte < min_dte or dte > max_dte: - continue - delta = _row_delta(row) - if delta is None or delta <= 0.0: - continue - bid, ask, mid = _row_bid_ask_mid(row) - if ask <= 0.0 or mid <= 0.0: - continue - spread_ratio = ((ask - bid) / mid) if bid > 0.0 else 0.0 - if spread_ratio > max_spread_ratio: - continue - strike = _as_float(row.get("strike"), default=0.0) - if strike <= 0.0: - continue - candidates.append((abs(dte - target_dte), abs(delta - target_delta), row, expiration, dte, delta, bid, ask, mid, strike)) - if not candidates: - _append_skip(intents_payload, recipe=recipe, underlier=underlier, reason="no_matching_option_contract") - return - - candidates.sort(key=lambda item: (item[0], item[1], item[9])) - _dte_gap, _delta_gap, row, expiration, dte, delta, bid, ask, mid, strike = candidates[0] - limit_price = round(min(ask, mid * 1.03), 2) - premium_budget = total_equity * _as_float(recipe_detail.get("premium_budget_ratio"), default=0.03) - quantity = int(premium_budget // (limit_price * multiplier)) if limit_price > 0.0 else 0 - existing_positions = _normalize_option_positions(ctx, underlier, "C") - - intents = list(intents_payload.get("intents") or ()) - if existing_positions: - roll_dte = int(_as_float(recipe_detail.get("roll_dte_months"), default=12.0) * 30.4375) - for position in existing_positions: - position_expiration = _parse_date(position.get("expiration") or position.get("lastTradeDateOrContractMonth")) - position_quantity = int(_as_float(position.get("quantity"), default=0.0)) - position_dte = (position_expiration - as_of).days if position_expiration is not None else 9999 - market_value = _as_float(position.get("market_value"), default=0.0) - cost_basis = _as_float(position.get("cost_basis"), default=0.0) - if position_quantity >= 2 and cost_basis > 0.0 and market_value >= cost_basis * 2.0 and bid > 0.0: - recover_quantity = max(1, int(cost_basis // (bid * multiplier)) + 1) - recover_quantity = min(position_quantity - 1, recover_quantity) - intents.append( - { - "intent_type": "single_leg_option", - "asset_class": "option", - "action": "sell_to_close", - "underlier": underlier, - "right": "C", - "expiration": str(position_expiration or position.get("expiration")), - "strike": _as_float(position.get("strike"), default=0.0), - "quantity": recover_quantity, - "order_type": "limit", - "limit_price": round(max(bid, 0.01), 2), - "time_in_force": "DAY", - "contract_multiplier": multiplier, - "reason": "recover_leaps_principal_after_2x", - } - ) - intents_payload["intents"] = intents - return - if position_dte <= roll_dte: - _append_skip(intents_payload, recipe=recipe, underlier=underlier, reason="roll_requires_existing_contract_quote") - return - _append_skip(intents_payload, recipe=recipe, underlier=underlier, reason="existing_position_held") - intents_payload["intents"] = intents - return - - if quantity < 1: - _append_skip(intents_payload, recipe=recipe, underlier=underlier, reason="contract_not_affordable") - return - - intents.append( - { - "intent_type": "single_leg_option", - "asset_class": "option", - "action": "buy_to_open", - "underlier": underlier, - "right": "C", - "expiration": expiration.isoformat(), - "strike": strike, - "quantity": quantity, - "order_type": "limit", - "limit_price": limit_price, - "time_in_force": "DAY", - "contract_multiplier": multiplier, - "max_notional_usd": round(quantity * limit_price * multiplier, 2), - "delta": round(delta, 4), - "dte": dte, - "reason": "open_leaps_growth", - } - ) - intents_payload["intents"] = intents - - -def _build_put_credit_spread_intents( - *, - recipe: str, - recipe_detail: Mapping[str, object], - ctx: StrategyContext, - total_equity: float, - base_diagnostics: Mapping[str, object], - intents_payload: dict[str, object], -) -> None: - underlier = str(recipe_detail.get("underlier") or "").strip().upper() - payload = _option_chain_payload(ctx, underlier) - gate_passed, gate_reason = _put_credit_spread_gate_passed(payload, recipe_detail, base_diagnostics) - if not gate_passed: - _append_skip(intents_payload, recipe=recipe, underlier=underlier, reason=gate_reason or "entry_gate_not_met") - return - rows = _chain_rows(payload) - spot = _chain_spot(payload, underlier, ctx) - if not rows or spot <= 0.0: - _append_skip(intents_payload, recipe=recipe, underlier=underlier, reason="missing_option_chain") - return - - as_of = _as_of_date(ctx) - target_dte = int(_as_float(recipe_detail.get("target_dte_days"), default=45.0)) - min_dte = int(_as_float(recipe_detail.get("min_dte_days"), default=25.0)) - max_dte = int(_as_float(recipe_detail.get("max_dte_days"), default=65.0)) - short_target = spot * (1.0 - _as_float(recipe_detail.get("short_put_otm_pct"), default=0.08)) - long_target = spot * (1.0 - _as_float(recipe_detail.get("long_put_otm_pct"), default=0.18)) - multiplier = int(_as_float(recipe_detail.get("contract_multiplier"), default=100.0)) - - puts_by_expiration: dict[date, list[Mapping[str, object]]] = {} - for row in rows: - if _normalize_right(row.get("right") or row.get("type")) != "P": - continue - expiration = _row_date(row) - if expiration is None: - continue - dte = (expiration - as_of).days - if min_dte <= dte <= max_dte: - puts_by_expiration.setdefault(expiration, []).append(row) - if not puts_by_expiration: - _append_skip(intents_payload, recipe=recipe, underlier=underlier, reason="no_matching_option_contract") - return - - expirations = sorted(puts_by_expiration, key=lambda expiry: abs((expiry - as_of).days - target_dte)) - best = None - for expiration in expirations: - rows_for_expiry = puts_by_expiration[expiration] - short_row = min(rows_for_expiry, key=lambda row: abs(_as_float(row.get("strike"), default=0.0) - short_target)) - long_candidates = [ - row for row in rows_for_expiry - if _as_float(row.get("strike"), default=0.0) < _as_float(short_row.get("strike"), default=0.0) - ] - if not long_candidates: - continue - long_row = min(long_candidates, key=lambda row: abs(_as_float(row.get("strike"), default=0.0) - long_target)) - short_bid, _short_ask, _short_mid = _row_bid_ask_mid(short_row) - _long_bid, long_ask, _long_mid = _row_bid_ask_mid(long_row) - short_strike = _as_float(short_row.get("strike"), default=0.0) - long_strike = _as_float(long_row.get("strike"), default=0.0) - net_credit = short_bid - long_ask - width = short_strike - long_strike - if short_bid > 0.0 and long_ask > 0.0 and net_credit > 0.05 and width > net_credit: - best = (expiration, short_row, long_row, short_strike, long_strike, round(net_credit, 2), width) - break - if best is None: - _append_skip(intents_payload, recipe=recipe, underlier=underlier, reason="no_viable_credit_spread") - return - - expiration, _short_row, _long_row, short_strike, long_strike, net_credit, width = best - max_loss = (width - net_credit) * multiplier - risk_budget = total_equity * _as_float(recipe_detail.get("max_loss_budget_ratio"), default=0.01) - quantity = int(risk_budget // max_loss) if max_loss > 0.0 else 0 - if quantity < 1: - _append_skip(intents_payload, recipe=recipe, underlier=underlier, reason="spread_not_affordable") - return - - intents = list(intents_payload.get("intents") or ()) - intents.append( - { - "intent_type": "multi_leg_option", - "asset_class": "option", - "action": "sell_to_open_put_credit_spread", - "underlier": underlier, - "expiration": expiration.isoformat(), - "quantity": quantity, - "order_type": "limit", - "limit_price": net_credit, - "time_in_force": "DAY", - "contract_multiplier": multiplier, - "max_loss_usd": round(quantity * max_loss, 2), - "net_credit_usd": round(quantity * net_credit * multiplier, 2), - "reason": "open_defined_risk_income_spread", - "legs": ( - { - "action": "sell_to_open", - "right": "P", - "strike": short_strike, - "expiration": expiration.isoformat(), - "ratio": 1, - }, - { - "action": "buy_to_open", - "right": "P", - "strike": long_strike, - "expiration": expiration.isoformat(), - "ratio": 1, - }, - ), - } - ) - intents_payload["intents"] = intents - - -def _attach_option_order_intents( - diagnostics: dict[str, object], - *, - recipe: str, - recipe_detail: Mapping[str, object], - ctx: StrategyContext, - total_equity: float, - base_diagnostics: Mapping[str, object], -) -> None: - intents_payload = dict( - diagnostics.get("option_order_intents") - or {"schema_version": "option_order_intents.v1", "intents": (), "skipped": ()} - ) - structure = str(recipe_detail.get("structure") or "").strip() - if structure == "long_call_leaps": - _build_long_call_intents( - recipe=recipe, - recipe_detail=recipe_detail, - ctx=ctx, - total_equity=total_equity, - base_diagnostics=base_diagnostics, - intents_payload=intents_payload, - ) - elif structure == "put_credit_spread": - _build_put_credit_spread_intents( - recipe=recipe, - recipe_detail=recipe_detail, - ctx=ctx, - total_equity=total_equity, - base_diagnostics=base_diagnostics, - intents_payload=intents_payload, - ) - else: - _append_skip( - intents_payload, - recipe=recipe, - underlier=str(recipe_detail.get("underlier") or "").strip().upper(), - reason="unsupported_option_overlay_structure", - ) - intents_payload["intents"] = tuple(intents_payload.get("intents") or ()) - intents_payload["skipped"] = tuple(intents_payload.get("skipped") or ()) - diagnostics["option_order_intents"] = intents_payload - diagnostics["option_order_intent_count"] = len(intents_payload["intents"]) - - -def build_option_overlay_diagnostics( - option_overlay_config: Mapping[str, object], - ctx: StrategyContext, - *, - base_diagnostics: Mapping[str, object] | None = None, -) -> dict[str, object]: - diagnostics: dict[str, object] = {} - if not option_overlay_config: - return diagnostics - - portfolio = ctx.portfolio - total_equity = ( - float(getattr(portfolio, "total_equity", 0.0) or 0.0) - if portfolio is not None - else 0.0 - ) - for family in ("growth", "income"): - prefix = f"option_{family}_overlay" - enabled = _as_bool(option_overlay_config.get(f"{prefix}_enabled"), default=False) - recipe = str(option_overlay_config.get(f"{prefix}_recipe") or "").strip() - start_usd = max(0.0, float(option_overlay_config.get(f"{prefix}_start_usd") or 0.0)) - if not enabled: - active = False - skip_reason = "disabled" - elif not recipe: - active = False - skip_reason = "missing_recipe" - elif portfolio is None: - active = False - skip_reason = "missing_portfolio" - elif total_equity < start_usd: - active = False - skip_reason = "below_start_usd" - else: - active = True - skip_reason = "" - - diagnostics[f"{prefix}_enabled"] = enabled - diagnostics[f"{prefix}_recipe"] = recipe - diagnostics[f"{prefix}_start_usd"] = start_usd - diagnostics[f"{prefix}_active"] = active - if skip_reason: - diagnostics[f"{prefix}_skip_reason"] = skip_reason - if recipe in OPTION_OVERLAY_RECIPE_DETAILS: - recipe_detail = _effective_option_recipe_detail( - family, - OPTION_OVERLAY_RECIPE_DETAILS[recipe], - option_overlay_config, - ) - diagnostics[f"{prefix}_recipe_detail"] = recipe_detail - else: - recipe_detail = {} - if active and recipe_detail: - _attach_option_order_intents( - diagnostics, - recipe=recipe, - recipe_detail=recipe_detail, - ctx=ctx, - total_equity=total_equity, - base_diagnostics=base_diagnostics or {}, - ) - return diagnostics - - def _portfolio_market_values(ctx: StrategyContext, symbols: tuple[str, ...]) -> dict[str, float]: market_values = {symbol: 0.0 for symbol in symbols} portfolio = ctx.portfolio diff --git a/src/us_equity_strategies/income_layer_defaults.py b/src/us_equity_strategies/income_layer_defaults.py index 745f6d7..8a541b2 100644 --- a/src/us_equity_strategies/income_layer_defaults.py +++ b/src/us_equity_strategies/income_layer_defaults.py @@ -9,6 +9,35 @@ SOXL_SOXX_TREND_INCOME_PROFILE = "soxl_soxx_trend_income" RUSSELL_TOP50_LEADER_ROTATION_PROFILE = "russell_top50_leader_rotation" +INCOME_LAYER_LIVE_VALIDATION_EVIDENCE: dict[str, dict[str, object]] = { + GLOBAL_ETF_ROTATION_PROFILE: { + "status": "live", + "evidence_status": "validated", + "research_doc": "docs/research/income_layer_design.zh-CN.md", + "summary": "Defensive ETF rotation income sleeve calibrated with drawdown-budget defaults.", + }, + TQQQ_GROWTH_INCOME_PROFILE: { + "status": "live", + "evidence_status": "validated", + "research_doc": "docs/research/income_layer_design.zh-CN.md", + "artifact": "UsEquitySnapshotPipelines/data/output/levered_income_layer_candidate_compare_2026-05-26/", + "summary": "TQQQ income sleeve selected from backtested drawdown-budget candidates.", + }, + SOXL_SOXX_TREND_INCOME_PROFILE: { + "status": "live", + "evidence_status": "validated", + "research_doc": "docs/research/income_layer_design.zh-CN.md", + "artifact": "UsEquitySnapshotPipelines/data/output/soxl_soxx_trend_income_archive_2026-05-04/summary.csv", + "summary": "SOXL/SOXX income sleeve validated by archived replay and later drawdown-budget calibration.", + }, + RUSSELL_TOP50_LEADER_ROTATION_PROFILE: { + "status": "live", + "evidence_status": "validated", + "research_doc": "docs/research/income_layer_design.zh-CN.md", + "summary": "Leader-rotation income sleeve uses the same validated drawdown-budget curve.", + }, +} + INCOME_LAYER_DEFAULT_CONFIGS: dict[str, dict[str, object]] = { GLOBAL_ETF_ROTATION_PROFILE: { "income_layer_enabled": True, @@ -39,8 +68,6 @@ "income_layer_base_drawdown_budget_ratio": 0.45, "income_layer_min_drawdown_budget_ratio": 0.25, "income_layer_drawdown_budget_decay_per_double": 0.05, - "income_layer_qqqi_weight": 0.02, - "income_layer_spyi_weight": 0.08, "income_layer_allocations": { "SCHD": 0.30, "DGRO": 0.20, @@ -60,8 +87,6 @@ "income_layer_base_drawdown_budget_ratio": 0.45, "income_layer_min_drawdown_budget_ratio": 0.25, "income_layer_drawdown_budget_decay_per_double": 0.05, - "income_layer_qqqi_weight": 0.01, - "income_layer_spyi_weight": 0.04, "income_layer_allocations": { "SCHD": 0.15, "DGRO": 0.10, @@ -96,6 +121,7 @@ def income_layer_default_config(profile: str) -> dict[str, object]: __all__ = [ "INCOME_LAYER_DEFAULT_CONFIGS", + "INCOME_LAYER_LIVE_VALIDATION_EVIDENCE", "INCOME_LAYER_RATIO_MODE", "income_layer_default_config", ] diff --git a/src/us_equity_strategies/manifests/__init__.py b/src/us_equity_strategies/manifests/__init__.py index 7fa541c..01afd93 100644 --- a/src/us_equity_strategies/manifests/__init__.py +++ b/src/us_equity_strategies/manifests/__init__.py @@ -4,6 +4,7 @@ from us_equity_strategies.ai_extensions import build_default_ai_extension_config from us_equity_strategies.income_layer_defaults import income_layer_default_config +from us_equity_strategies.option_overlay import option_overlay_default_config GLOBAL_ETF_CONFIDENCE_VOL_GATE_PROFILE = "global_etf_confidence_vol_gate" RUSSELL_TOP50_LEADER_ROTATION_PROFILE = "russell_top50_leader_rotation" @@ -79,6 +80,7 @@ def _manifest( "confidence_volatility_window": 126, "confidence_volatility_max_ratio": 1.3, **income_layer_default_config("global_etf_rotation"), + **option_overlay_default_config("global_etf_rotation"), "market_regime_control_enabled": True, "market_regime_control_apply_risk_reduced": False, "market_regime_control_apply_risk_off": False, @@ -96,16 +98,11 @@ def _manifest( default_config={ "benchmark_symbol": "QQQ", "managed_symbols": ("TQQQ", "QQQM", "BOXX", "SCHD", "DGRO", "SGOV", "SPYI", "QQQI"), - "income_threshold_usd": 250000.0, - "qqqi_income_ratio": 0.10, "cash_reserve_ratio": 0.02, "rebalance_threshold_ratio": 0.01, "execution_cash_reserve_ratio": 0.0, **income_layer_default_config("tqqq_growth_income"), - "option_growth_overlay_enabled": True, - "option_growth_overlay_recipe": "tqqq_leaps_growth_v1", - "option_growth_overlay_start_usd": 250000.0, - "option_growth_overlay_nav_budget_ratio": 0.03, + **option_overlay_default_config("tqqq_growth_income"), "attack_allocation_mode": "fixed_qqq_tqqq_pullback", "dual_drive_qqq_weight": 0.45, "dual_drive_tqqq_weight": 0.45, @@ -153,10 +150,7 @@ def _manifest( "min_trade_floor": 100.0, "rebalance_threshold_ratio": 0.01, **income_layer_default_config("soxl_soxx_trend_income"), - "option_income_overlay_enabled": True, - "option_income_overlay_recipe": "soxx_put_credit_spread_income_v1", - "option_income_overlay_start_usd": 1000000.0, - "option_income_overlay_nav_risk_ratio": 0.01, + **option_overlay_default_config("soxl_soxx_trend_income"), "trend_entry_buffer": 0.08, "trend_mid_buffer": 0.06, "trend_exit_buffer": 0.02, @@ -222,10 +216,7 @@ def _manifest( "runtime_execution_window_trading_days": 3, "execution_cash_reserve_ratio": 0.0, **income_layer_default_config(RUSSELL_TOP50_LEADER_ROTATION_PROFILE), - "option_growth_overlay_enabled": True, - "option_growth_overlay_recipe": "qqq_leaps_growth_v1", - "option_growth_overlay_start_usd": 1000000.0, - "option_growth_overlay_nav_budget_ratio": 0.03, + **option_overlay_default_config(RUSSELL_TOP50_LEADER_ROTATION_PROFILE), "market_regime_control_enabled": True, "market_regime_control_apply_risk_reduced": False, "market_regime_control_apply_risk_off": False, diff --git a/src/us_equity_strategies/option_overlay.py b/src/us_equity_strategies/option_overlay.py new file mode 100644 index 0000000..2aa92d5 --- /dev/null +++ b/src/us_equity_strategies/option_overlay.py @@ -0,0 +1,884 @@ +from __future__ import annotations + +from collections.abc import Mapping +from copy import deepcopy +from datetime import date, datetime + +from quant_platform_kit.strategy_contracts import StrategyContext + + +OPTION_OVERLAY_CONFIG_KEYS = { + "option_overlay_enabled", + "option_growth_overlay_enabled", + "option_growth_overlay_recipe", + "option_growth_overlay_start_usd", + "option_growth_overlay_nav_budget_ratio", + "option_income_overlay_enabled", + "option_income_overlay_recipe", + "option_income_overlay_start_usd", + "option_income_overlay_nav_risk_ratio", +} + +OPTION_OVERLAY_RECIPE_DETAILS = { + "tqqq_leaps_growth_v1": { + "structure": "long_call_leaps", + "underlier": "TQQQ", + "premium_budget_ratio": 0.03, + "target_delta": 0.75, + "target_dte_months": 24, + "roll_dte_months": 12, + "contract_multiplier": 100, + "min_dte_days": 540, + "max_dte_days": 930, + "max_bid_ask_spread_ratio": 0.12, + "entry_gate": "core_risk_on_and_qqq_above_200dma_and_qqq_63d_momentum_positive", + "principal_take_profit": "sell_enough_contracts_to_recover_cost_after_2x_when_contract_count_allows", + }, + "qqq_leaps_growth_v1": { + "structure": "long_call_leaps", + "underlier": "QQQ", + "premium_budget_ratio": 0.03, + "target_delta": 0.75, + "target_dte_months": 24, + "roll_dte_months": 12, + "contract_multiplier": 100, + "min_dte_days": 540, + "max_dte_days": 930, + "max_bid_ask_spread_ratio": 0.12, + "entry_gate": "qqq_above_200dma_and_qqq_63d_momentum_positive", + "principal_take_profit": "sell_enough_contracts_to_recover_cost_after_2x_when_contract_count_allows", + }, + "spy_leaps_growth_v1": { + "structure": "long_call_leaps", + "underlier": "SPY", + "premium_budget_ratio": 0.03, + "target_delta": 0.75, + "target_dte_months": 24, + "roll_dte_months": 12, + "contract_multiplier": 100, + "min_dte_days": 540, + "max_dte_days": 930, + "max_bid_ask_spread_ratio": 0.10, + "entry_gate": "spy_above_200dma_and_spy_63d_momentum_positive", + "principal_take_profit": "sell_enough_contracts_to_recover_cost_after_2x_when_contract_count_allows", + }, + "soxx_put_credit_spread_income_v1": { + "structure": "put_credit_spread", + "underlier": "SOXX", + "target_dte_days": 45, + "short_put_otm_pct": 0.08, + "long_put_otm_pct": 0.18, + "max_loss_budget_ratio": 0.01, + "contract_multiplier": 100, + "min_dte_days": 25, + "max_dte_days": 65, + "max_iv_rank": 0.80, + "entry_gate": "soxx_trend_positive_and_iv_rank_not_extreme", + }, +} + +OPTION_OVERLAY_RESEARCH_CANDIDATES = { + "tqqq_leaps_growth_v1": { + "status": "research", + "promotion_evidence": False, + "reason": "requires historical option-chain bid/ask validation before live defaults", + }, + "qqq_leaps_growth_v1": { + "status": "research", + "promotion_evidence": False, + "reason": "proxy backtest only; requires historical option-chain bid/ask validation", + }, + "spy_leaps_growth_v1": { + "status": "research", + "promotion_evidence": False, + "reason": "proxy backtest only; requires historical option-chain bid/ask validation", + }, + "soxx_put_credit_spread_income_v1": { + "status": "research", + "promotion_evidence": False, + "reason": "defined-risk income candidate; requires historical option-chain spread validation", + }, +} + + +OPTION_OVERLAY_DEFAULT_CONFIGS = { + "global_etf_rotation": { + "option_overlay_enabled": True, + "option_growth_overlay_enabled": True, + "option_growth_overlay_recipe": "spy_leaps_growth_v1", + "option_growth_overlay_start_usd": 500000.0, + "option_growth_overlay_nav_budget_ratio": 0.015, + "option_income_overlay_enabled": False, + }, + "tqqq_growth_income": { + "option_overlay_enabled": True, + "option_growth_overlay_enabled": True, + "option_growth_overlay_recipe": "tqqq_leaps_growth_v1", + "option_growth_overlay_start_usd": 250000.0, + "option_growth_overlay_nav_budget_ratio": 0.03, + "option_income_overlay_enabled": False, + }, + "soxl_soxx_trend_income": { + "option_overlay_enabled": True, + "option_growth_overlay_enabled": False, + "option_income_overlay_enabled": True, + "option_income_overlay_recipe": "soxx_put_credit_spread_income_v1", + "option_income_overlay_start_usd": 150000.0, + "option_income_overlay_nav_risk_ratio": 0.01, + }, + "russell_top50_leader_rotation": { + "option_overlay_enabled": True, + "option_growth_overlay_enabled": True, + "option_growth_overlay_recipe": "spy_leaps_growth_v1", + "option_growth_overlay_start_usd": 300000.0, + "option_growth_overlay_nav_budget_ratio": 0.015, + "option_income_overlay_enabled": False, + }, +} + + +def option_overlay_default_config(profile: str) -> dict[str, object]: + normalized = str(profile or "").strip().lower() + return deepcopy(OPTION_OVERLAY_DEFAULT_CONFIGS.get(normalized, {})) + + +def _recipe_live_allowed(recipe: str) -> tuple[bool, str, str]: + candidate = OPTION_OVERLAY_RESEARCH_CANDIDATES.get(recipe) + if not candidate: + return False, "missing_promotion_record", "" + status = str(candidate.get("status") or "").strip().lower() + if status != "live" or candidate.get("promotion_evidence") is not True: + return False, "research_only_recipe", status + return True, "", status + + +def _as_bool(value: object, *, default: bool = False) -> bool: + if value is None: + return default + if isinstance(value, bool): + return value + if isinstance(value, str): + normalized = value.strip().lower() + if normalized in {"1", "true", "yes", "y", "on"}: + return True + if normalized in {"0", "false", "no", "n", "off"}: + return False + return bool(value) + + +def _as_float(value: object, *, default: float = 0.0) -> float: + try: + return float(value) + except (TypeError, ValueError): + return default + + +def _clamped_ratio(value: object, *, default: float, upper: float) -> float: + return max(0.0, min(float(upper), _as_float(value, default=default))) + + +def _effective_option_recipe_detail( + family: str, + recipe_detail: Mapping[str, object], + option_overlay_config: Mapping[str, object], +) -> dict[str, object]: + effective = dict(recipe_detail) + if family == "growth": + default = _as_float(recipe_detail.get("premium_budget_ratio"), default=0.03) + effective["premium_budget_ratio"] = _clamped_ratio( + option_overlay_config.get("option_growth_overlay_nav_budget_ratio"), + default=default, + upper=0.10, + ) + elif family == "income": + default = _as_float(recipe_detail.get("max_loss_budget_ratio"), default=0.01) + effective["max_loss_budget_ratio"] = _clamped_ratio( + option_overlay_config.get("option_income_overlay_nav_risk_ratio"), + default=default, + upper=0.02, + ) + return effective + + +def _parse_date(value: object) -> date | None: + if value is None or value == "": + return None + if isinstance(value, datetime): + return value.date() + if isinstance(value, date): + return value + text = str(value).strip() + if not text: + return None + if len(text) == 8 and text.isdigit(): + text = f"{text[:4]}-{text[4:6]}-{text[6:]}" + try: + return datetime.fromisoformat(text[:10]).date() + except ValueError: + return None + + +def _as_of_date(ctx: StrategyContext) -> date: + parsed = _parse_date(ctx.as_of) + return parsed or datetime.utcnow().date() + + +def _normalize_right(value: object) -> str: + text = str(value or "").strip().upper() + if text in {"CALL", "C"}: + return "C" + if text in {"PUT", "P"}: + return "P" + return text + + +def _option_chain_payload(ctx: StrategyContext, underlier: str) -> object: + underlier = str(underlier or "").strip().upper() + for source in (ctx.market_data, ctx.runtime_config): + payload = source.get("option_chains") if isinstance(source, Mapping) else None + if isinstance(payload, Mapping): + direct = payload.get(underlier) or payload.get(underlier.lower()) + if direct is not None: + return direct + payload = source.get("option_chain") if isinstance(source, Mapping) else None + if isinstance(payload, Mapping): + payload_underlier = str( + payload.get("underlier") or payload.get("symbol") or "" + ).strip().upper() + if payload_underlier == underlier: + return payload + elif payload: + return payload + return None + + +def _chain_rows(payload: object) -> tuple[Mapping[str, object], ...]: + if payload is None: + return () + if isinstance(payload, Mapping): + for key in ("contracts", "options", "rows", "chain"): + rows = payload.get(key) + if rows is not None: + return tuple(row for row in rows if isinstance(row, Mapping)) + return () + try: + return tuple(row for row in payload if isinstance(row, Mapping)) # type: ignore[union-attr] + except TypeError: + return () + + +def _chain_spot(payload: object, underlier: str, ctx: StrategyContext) -> float: + if isinstance(payload, Mapping): + for key in ("spot", "underlier_price", "underlying_price", "last_price"): + price = _as_float(payload.get(key), default=0.0) + if price > 0.0: + return price + quote = ctx.market_data.get(str(underlier).strip().upper()) + if isinstance(quote, Mapping): + for key in ("last_price", "last", "close", "price"): + price = _as_float(quote.get(key), default=0.0) + if price > 0.0: + return price + return 0.0 + + +def _indicator_payload(ctx: StrategyContext, symbol: str) -> Mapping[str, object]: + symbol = str(symbol or "").strip().upper() + sources = (ctx.market_data, ctx.runtime_config) + for source in sources: + if not isinstance(source, Mapping): + continue + for key in ("underlier_indicators", "market_indicators", "derived_indicators", "indicators"): + payload = source.get(key) + if isinstance(payload, Mapping): + direct = payload.get(symbol) or payload.get(symbol.lower()) + if isinstance(direct, Mapping): + return direct + direct = source.get(symbol) or source.get(symbol.lower()) + if isinstance(direct, Mapping): + return direct + return {} + + +def _optional_bool_indicator(payload: Mapping[str, object], *keys: str) -> bool | None: + for key in keys: + if key not in payload: + continue + value = payload.get(key) + if value is None or value == "": + continue + return _as_bool(value, default=False) + return None + + +def _optional_float_indicator(payload: Mapping[str, object], *keys: str) -> float | None: + for key in keys: + if key not in payload: + continue + value = payload.get(key) + if value is None or value == "": + continue + return _as_float(value, default=0.0) + return None + + +def _row_date(row: Mapping[str, object]) -> date | None: + for key in ("expiration", "expiry", "lastTradeDateOrContractMonth", "last_trade_date"): + parsed = _parse_date(row.get(key)) + if parsed is not None: + return parsed + return None + + +def _row_delta(row: Mapping[str, object]) -> float | None: + for key in ("delta", "model_delta"): + if key in row: + return _as_float(row.get(key), default=0.0) + greeks = row.get("greeks") + if isinstance(greeks, Mapping) and "delta" in greeks: + return _as_float(greeks.get("delta"), default=0.0) + return None + + +def _row_bid_ask_mid(row: Mapping[str, object]) -> tuple[float, float, float]: + bid = _as_float(row.get("bid") or row.get("bid_price"), default=0.0) + ask = _as_float(row.get("ask") or row.get("ask_price"), default=0.0) + raw_mid = _as_float(row.get("mid") or row.get("mark") or row.get("last"), default=0.0) + if bid > 0.0 and ask > 0.0: + return bid, ask, (bid + ask) / 2.0 + if raw_mid > 0.0 and ask <= 0.0: + ask = raw_mid + if raw_mid > 0.0: + return bid, ask, raw_mid + return bid, ask, ask if ask > 0.0 else bid + + +def _normalize_option_positions( + ctx: StrategyContext, + underlier: str, + right: str, +) -> tuple[Mapping[str, object], ...]: + portfolio = ctx.portfolio + metadata = getattr(portfolio, "metadata", {}) if portfolio is not None else {} + rows = metadata.get("option_positions") if isinstance(metadata, Mapping) else () + normalized = [] + for row in rows or (): + if not isinstance(row, Mapping): + continue + row_underlier = str(row.get("underlier") or row.get("symbol") or "").strip().upper() + row_right = _normalize_right(row.get("right") or row.get("put_call")) + quantity = _as_float(row.get("quantity"), default=0.0) + if row_underlier == underlier and row_right == right and quantity > 0.0: + normalized.append(row) + return tuple(normalized) + + +def _long_call_entry_gate_passed( + recipe: str, + underlier: str, + ctx: StrategyContext, + base_diagnostics: Mapping[str, object], +) -> tuple[bool, str | None]: + signal_context = base_diagnostics.get("notification_context") + if isinstance(signal_context, Mapping): + raw_signal = signal_context.get("signal") + if isinstance(raw_signal, Mapping): + state = str(raw_signal.get("state") or "").strip().lower() + if state: + return state in {"entry", "hold"}, None if state in {"entry", "hold"} else "entry_gate_not_met" + regime = str(base_diagnostics.get("regime") or "").strip().lower() + if recipe in {"qqq_leaps_growth_v1", "spy_leaps_growth_v1"} and regime: + if regime != "risk_on": + return False, "entry_gate_not_met" + indicators = _indicator_payload(ctx, underlier) + if indicators: + above_long_trend = _optional_bool_indicator( + indicators, + "above_200dma", + "above_ma200", + "sma200_pass", + "ma200_pass", + "trend_200d_pass", + ) + if above_long_trend is False: + return False, "entry_gate_not_met" + momentum = _optional_float_indicator( + indicators, + "momentum_63d", + "momentum_3m", + "return_63d", + "roc_63d", + ) + if momentum is not None and momentum <= 0.0: + return False, "entry_gate_not_met" + return True, None + + +def _put_credit_spread_gate_passed( + payload: object, + recipe_detail: Mapping[str, object], + base_diagnostics: Mapping[str, object], +) -> tuple[bool, str | None]: + blend_tier = str(base_diagnostics.get("blend_tier") or "").strip().lower() + if blend_tier and blend_tier not in {"full", "mid"}: + return False, "entry_gate_not_met" + iv_rank = None + if isinstance(payload, Mapping): + for key in ("iv_rank", "implied_volatility_rank"): + if key in payload: + iv_rank = _as_float(payload.get(key), default=-1.0) + break + if iv_rank is not None and iv_rank > _as_float(recipe_detail.get("max_iv_rank"), default=0.80): + return False, "iv_rank_too_high" + return True, None + + +def _append_skip( + intents_payload: dict[str, object], + *, + recipe: str, + reason: str, + underlier: str, +) -> None: + skipped = list(intents_payload.get("skipped") or ()) + skipped.append({"recipe": recipe, "underlier": underlier, "reason": reason}) + intents_payload["skipped"] = skipped + + +def _build_long_call_intents( + *, + recipe: str, + recipe_detail: Mapping[str, object], + ctx: StrategyContext, + total_equity: float, + base_diagnostics: Mapping[str, object], + intents_payload: dict[str, object], +) -> None: + underlier = str(recipe_detail.get("underlier") or "").strip().upper() + gate_passed, gate_reason = _long_call_entry_gate_passed(recipe, underlier, ctx, base_diagnostics) + if not gate_passed: + _append_skip( + intents_payload, + recipe=recipe, + underlier=underlier, + reason=gate_reason or "entry_gate_not_met", + ) + return + + payload = _option_chain_payload(ctx, underlier) + rows = _chain_rows(payload) + if not rows: + _append_skip( + intents_payload, + recipe=recipe, + underlier=underlier, + reason="missing_option_chain", + ) + return + + as_of = _as_of_date(ctx) + target_dte = int(_as_float(recipe_detail.get("target_dte_months"), default=24.0) * 30.4375) + target_delta = _as_float(recipe_detail.get("target_delta"), default=0.75) + min_dte = int(_as_float(recipe_detail.get("min_dte_days"), default=540.0)) + max_dte = int(_as_float(recipe_detail.get("max_dte_days"), default=930.0)) + max_spread_ratio = _as_float(recipe_detail.get("max_bid_ask_spread_ratio"), default=0.12) + multiplier = int(_as_float(recipe_detail.get("contract_multiplier"), default=100.0)) + candidates = [] + for row in rows: + if _normalize_right(row.get("right") or row.get("type")) != "C": + continue + expiration = _row_date(row) + if expiration is None: + continue + dte = (expiration - as_of).days + if dte < min_dte or dte > max_dte: + continue + delta = _row_delta(row) + if delta is None or delta <= 0.0: + continue + bid, ask, mid = _row_bid_ask_mid(row) + if ask <= 0.0 or mid <= 0.0: + continue + spread_ratio = ((ask - bid) / mid) if bid > 0.0 else 0.0 + if spread_ratio > max_spread_ratio: + continue + strike = _as_float(row.get("strike"), default=0.0) + if strike <= 0.0: + continue + candidates.append( + (abs(dte - target_dte), abs(delta - target_delta), row, expiration, dte, delta, bid, ask, mid, strike) + ) + if not candidates: + _append_skip( + intents_payload, + recipe=recipe, + underlier=underlier, + reason="no_matching_option_contract", + ) + return + + candidates.sort(key=lambda item: (item[0], item[1], item[9])) + _dte_gap, _delta_gap, row, expiration, dte, delta, bid, ask, mid, strike = candidates[0] + limit_price = round(min(ask, mid * 1.03), 2) + premium_budget = total_equity * _as_float(recipe_detail.get("premium_budget_ratio"), default=0.03) + quantity = int(premium_budget // (limit_price * multiplier)) if limit_price > 0.0 else 0 + existing_positions = _normalize_option_positions(ctx, underlier, "C") + + intents = list(intents_payload.get("intents") or ()) + if existing_positions: + roll_dte = int(_as_float(recipe_detail.get("roll_dte_months"), default=12.0) * 30.4375) + for position in existing_positions: + position_expiration = _parse_date( + position.get("expiration") or position.get("lastTradeDateOrContractMonth") + ) + position_quantity = int(_as_float(position.get("quantity"), default=0.0)) + position_dte = (position_expiration - as_of).days if position_expiration is not None else 9999 + market_value = _as_float(position.get("market_value"), default=0.0) + cost_basis = _as_float(position.get("cost_basis"), default=0.0) + if position_quantity >= 2 and cost_basis > 0.0 and market_value >= cost_basis * 2.0 and bid > 0.0: + recover_quantity = max(1, int(cost_basis // (bid * multiplier)) + 1) + recover_quantity = min(position_quantity - 1, recover_quantity) + intents.append( + { + "intent_type": "single_leg_option", + "asset_class": "option", + "action": "sell_to_close", + "underlier": underlier, + "right": "C", + "expiration": str(position_expiration or position.get("expiration")), + "strike": _as_float(position.get("strike"), default=0.0), + "quantity": recover_quantity, + "order_type": "limit", + "limit_price": round(max(bid, 0.01), 2), + "time_in_force": "DAY", + "contract_multiplier": multiplier, + "reason": "recover_leaps_principal_after_2x", + } + ) + intents_payload["intents"] = intents + return + if position_dte <= roll_dte: + _append_skip( + intents_payload, + recipe=recipe, + underlier=underlier, + reason="roll_requires_existing_contract_quote", + ) + return + _append_skip( + intents_payload, + recipe=recipe, + underlier=underlier, + reason="existing_position_held", + ) + intents_payload["intents"] = intents + return + + if quantity < 1: + _append_skip( + intents_payload, + recipe=recipe, + underlier=underlier, + reason="contract_not_affordable", + ) + return + + intents.append( + { + "intent_type": "single_leg_option", + "asset_class": "option", + "action": "buy_to_open", + "underlier": underlier, + "right": "C", + "expiration": expiration.isoformat(), + "strike": strike, + "quantity": quantity, + "order_type": "limit", + "limit_price": limit_price, + "time_in_force": "DAY", + "contract_multiplier": multiplier, + "max_notional_usd": round(quantity * limit_price * multiplier, 2), + "delta": round(delta, 4), + "dte": dte, + "reason": "open_leaps_growth", + } + ) + intents_payload["intents"] = intents + + +def _build_put_credit_spread_intents( + *, + recipe: str, + recipe_detail: Mapping[str, object], + ctx: StrategyContext, + total_equity: float, + base_diagnostics: Mapping[str, object], + intents_payload: dict[str, object], +) -> None: + underlier = str(recipe_detail.get("underlier") or "").strip().upper() + payload = _option_chain_payload(ctx, underlier) + gate_passed, gate_reason = _put_credit_spread_gate_passed(payload, recipe_detail, base_diagnostics) + if not gate_passed: + _append_skip( + intents_payload, + recipe=recipe, + underlier=underlier, + reason=gate_reason or "entry_gate_not_met", + ) + return + rows = _chain_rows(payload) + spot = _chain_spot(payload, underlier, ctx) + if not rows or spot <= 0.0: + _append_skip( + intents_payload, + recipe=recipe, + underlier=underlier, + reason="missing_option_chain", + ) + return + + as_of = _as_of_date(ctx) + target_dte = int(_as_float(recipe_detail.get("target_dte_days"), default=45.0)) + min_dte = int(_as_float(recipe_detail.get("min_dte_days"), default=25.0)) + max_dte = int(_as_float(recipe_detail.get("max_dte_days"), default=65.0)) + short_target = spot * (1.0 - _as_float(recipe_detail.get("short_put_otm_pct"), default=0.08)) + long_target = spot * (1.0 - _as_float(recipe_detail.get("long_put_otm_pct"), default=0.18)) + multiplier = int(_as_float(recipe_detail.get("contract_multiplier"), default=100.0)) + + puts_by_expiration: dict[date, list[Mapping[str, object]]] = {} + for row in rows: + if _normalize_right(row.get("right") or row.get("type")) != "P": + continue + expiration = _row_date(row) + if expiration is None: + continue + dte = (expiration - as_of).days + if min_dte <= dte <= max_dte: + puts_by_expiration.setdefault(expiration, []).append(row) + if not puts_by_expiration: + _append_skip( + intents_payload, + recipe=recipe, + underlier=underlier, + reason="no_matching_option_contract", + ) + return + + expirations = sorted(puts_by_expiration, key=lambda expiry: abs((expiry - as_of).days - target_dte)) + best = None + for expiration in expirations: + rows_for_expiry = puts_by_expiration[expiration] + short_row = min(rows_for_expiry, key=lambda row: abs(_as_float(row.get("strike"), default=0.0) - short_target)) + long_candidates = [ + row + for row in rows_for_expiry + if _as_float(row.get("strike"), default=0.0) < _as_float(short_row.get("strike"), default=0.0) + ] + if not long_candidates: + continue + long_row = min(long_candidates, key=lambda row: abs(_as_float(row.get("strike"), default=0.0) - long_target)) + short_bid, _short_ask, _short_mid = _row_bid_ask_mid(short_row) + _long_bid, long_ask, _long_mid = _row_bid_ask_mid(long_row) + short_strike = _as_float(short_row.get("strike"), default=0.0) + long_strike = _as_float(long_row.get("strike"), default=0.0) + net_credit = short_bid - long_ask + width = short_strike - long_strike + if short_bid > 0.0 and long_ask > 0.0 and net_credit > 0.05 and width > net_credit: + best = (expiration, short_row, long_row, short_strike, long_strike, round(net_credit, 2), width) + break + if best is None: + _append_skip( + intents_payload, + recipe=recipe, + underlier=underlier, + reason="no_viable_credit_spread", + ) + return + + expiration, _short_row, _long_row, short_strike, long_strike, net_credit, width = best + max_loss = (width - net_credit) * multiplier + risk_budget = total_equity * _as_float(recipe_detail.get("max_loss_budget_ratio"), default=0.01) + quantity = int(risk_budget // max_loss) if max_loss > 0.0 else 0 + if quantity < 1: + _append_skip( + intents_payload, + recipe=recipe, + underlier=underlier, + reason="spread_not_affordable", + ) + return + + intents = list(intents_payload.get("intents") or ()) + intents.append( + { + "intent_type": "multi_leg_option", + "asset_class": "option", + "action": "sell_to_open_put_credit_spread", + "underlier": underlier, + "expiration": expiration.isoformat(), + "quantity": quantity, + "order_type": "limit", + "limit_price": net_credit, + "time_in_force": "DAY", + "contract_multiplier": multiplier, + "max_loss_usd": round(quantity * max_loss, 2), + "net_credit_usd": round(quantity * net_credit * multiplier, 2), + "reason": "open_defined_risk_income_spread", + "legs": ( + { + "action": "sell_to_open", + "right": "P", + "strike": short_strike, + "expiration": expiration.isoformat(), + "ratio": 1, + }, + { + "action": "buy_to_open", + "right": "P", + "strike": long_strike, + "expiration": expiration.isoformat(), + "ratio": 1, + }, + ), + } + ) + intents_payload["intents"] = intents + + +def _attach_option_order_intents( + diagnostics: dict[str, object], + *, + recipe: str, + recipe_detail: Mapping[str, object], + ctx: StrategyContext, + total_equity: float, + base_diagnostics: Mapping[str, object], +) -> None: + intents_payload = dict( + diagnostics.get("option_order_intents") + or {"schema_version": "option_order_intents.v1", "intents": (), "skipped": ()} + ) + structure = str(recipe_detail.get("structure") or "").strip() + if structure == "long_call_leaps": + _build_long_call_intents( + recipe=recipe, + recipe_detail=recipe_detail, + ctx=ctx, + total_equity=total_equity, + base_diagnostics=base_diagnostics, + intents_payload=intents_payload, + ) + elif structure == "put_credit_spread": + _build_put_credit_spread_intents( + recipe=recipe, + recipe_detail=recipe_detail, + ctx=ctx, + total_equity=total_equity, + base_diagnostics=base_diagnostics, + intents_payload=intents_payload, + ) + else: + _append_skip( + intents_payload, + recipe=recipe, + underlier=str(recipe_detail.get("underlier") or "").strip().upper(), + reason="unsupported_option_overlay_structure", + ) + intents_payload["intents"] = tuple(intents_payload.get("intents") or ()) + intents_payload["skipped"] = tuple(intents_payload.get("skipped") or ()) + diagnostics["option_order_intents"] = intents_payload + diagnostics["option_order_intent_count"] = len(intents_payload["intents"]) + + +def build_option_overlay_diagnostics( + option_overlay_config: Mapping[str, object], + ctx: StrategyContext, + *, + base_diagnostics: Mapping[str, object] | None = None, +) -> dict[str, object]: + diagnostics: dict[str, object] = {} + if not option_overlay_config: + return diagnostics + + option_overlay_enabled = _as_bool(option_overlay_config.get("option_overlay_enabled"), default=True) + diagnostics["option_overlay_enabled"] = option_overlay_enabled + portfolio = ctx.portfolio + total_equity = ( + float(getattr(portfolio, "total_equity", 0.0) or 0.0) + if portfolio is not None + else 0.0 + ) + for family in ("growth", "income"): + prefix = f"option_{family}_overlay" + enabled = _as_bool(option_overlay_config.get(f"{prefix}_enabled"), default=False) + recipe = str(option_overlay_config.get(f"{prefix}_recipe") or "").strip() + start_usd = max(0.0, float(option_overlay_config.get(f"{prefix}_start_usd") or 0.0)) + live_allowed = False + promotion_status = "" + if not option_overlay_enabled: + active = False + skip_reason = "option_overlay_disabled" + elif not enabled: + active = False + skip_reason = "disabled" + elif not recipe: + active = False + skip_reason = "missing_recipe" + elif not (live_gate := _recipe_live_allowed(recipe))[0]: + active = False + skip_reason = live_gate[1] + promotion_status = live_gate[2] + elif portfolio is None: + active = False + skip_reason = "missing_portfolio" + live_allowed = True + promotion_status = live_gate[2] + elif total_equity < start_usd: + active = False + skip_reason = "below_start_usd" + live_allowed = True + promotion_status = live_gate[2] + else: + active = True + skip_reason = "" + live_allowed = True + promotion_status = live_gate[2] + + diagnostics[f"{prefix}_enabled"] = enabled + diagnostics[f"{prefix}_recipe"] = recipe + diagnostics[f"{prefix}_start_usd"] = start_usd + diagnostics[f"{prefix}_live_allowed"] = live_allowed + if promotion_status: + diagnostics[f"{prefix}_promotion_status"] = promotion_status + diagnostics[f"{prefix}_active"] = active + if skip_reason: + diagnostics[f"{prefix}_skip_reason"] = skip_reason + if recipe in OPTION_OVERLAY_RECIPE_DETAILS: + recipe_detail = _effective_option_recipe_detail( + family, + OPTION_OVERLAY_RECIPE_DETAILS[recipe], + option_overlay_config, + ) + diagnostics[f"{prefix}_recipe_detail"] = recipe_detail + else: + recipe_detail = {} + if active and recipe_detail: + _attach_option_order_intents( + diagnostics, + recipe=recipe, + recipe_detail=recipe_detail, + ctx=ctx, + total_equity=total_equity, + base_diagnostics=base_diagnostics or {}, + ) + return diagnostics + + +__all__ = [ + "OPTION_OVERLAY_CONFIG_KEYS", + "OPTION_OVERLAY_DEFAULT_CONFIGS", + "OPTION_OVERLAY_RESEARCH_CANDIDATES", + "OPTION_OVERLAY_RECIPE_DETAILS", + "build_option_overlay_diagnostics", + "option_overlay_default_config", +] diff --git a/src/us_equity_strategies/strategies/soxl_soxx_trend_income.py b/src/us_equity_strategies/strategies/soxl_soxx_trend_income.py index b8da01e..dc66001 100644 --- a/src/us_equity_strategies/strategies/soxl_soxx_trend_income.py +++ b/src/us_equity_strategies/strategies/soxl_soxx_trend_income.py @@ -399,8 +399,8 @@ def build_rebalance_plan( rebalance_threshold_ratio, income_layer_start_usd, income_layer_max_ratio, - income_layer_qqqi_weight, - income_layer_spyi_weight, + income_layer_qqqi_weight=None, + income_layer_spyi_weight=None, income_layer_allocations=None, income_layer_enabled=True, income_layer_activation_band_ratio=0.0, diff --git a/src/us_equity_strategies/strategies/tqqq_growth_income.py b/src/us_equity_strategies/strategies/tqqq_growth_income.py index 0420bf8..b2d7a4c 100644 --- a/src/us_equity_strategies/strategies/tqqq_growth_income.py +++ b/src/us_equity_strategies/strategies/tqqq_growth_income.py @@ -500,11 +500,11 @@ def build_rebalance_plan( *, signal_text_fn, translator, - income_threshold_usd, - qqqi_income_ratio, cash_reserve_ratio, - cash_reserve_floor_usd=0.0, rebalance_threshold_ratio, + cash_reserve_floor_usd=0.0, + income_threshold_usd=None, + qqqi_income_ratio=None, income_layer_start_usd=None, income_layer_max_ratio=None, income_layer_qqqi_weight=None, @@ -592,7 +592,13 @@ def build_rebalance_plan( if not isinstance(snapshot_metadata, Mapping): snapshot_metadata = {} - layer_start = income_threshold_usd if income_layer_start_usd is None else income_layer_start_usd + layer_start = ( + 0.0 + if income_layer_start_usd is None and income_threshold_usd is None + else income_threshold_usd + if income_layer_start_usd is None + else income_layer_start_usd + ) layer_max_ratio = 0.60 if income_layer_max_ratio is None else income_layer_max_ratio income_layer_plan = build_income_layer_plan( total_equity_usd=total_equity, diff --git a/tests/test_catalog.py b/tests/test_catalog.py index 6b41563..18d591a 100644 --- a/tests/test_catalog.py +++ b/tests/test_catalog.py @@ -26,6 +26,12 @@ get_strategy_platform_compatibility_map, resolve_canonical_profile, ) +from us_equity_strategies.income_layer_defaults import INCOME_LAYER_LIVE_VALIDATION_EVIDENCE +from us_equity_strategies.option_overlay import ( + OPTION_OVERLAY_CONFIG_KEYS, + OPTION_OVERLAY_DEFAULT_CONFIGS, + OPTION_OVERLAY_RESEARCH_CANDIDATES, +) class CatalogTest(unittest.TestCase): @@ -233,34 +239,66 @@ def test_metadata_map_exposes_display_names_and_roles(self): FULL_SHARED_PLATFORM_MATRIX, ) - def test_option_overlay_defaults_are_scoped_to_supported_profiles(self): - tqqq = get_strategy_definition(TQQQ_GROWTH_INCOME_PROFILE).default_config - self.assertIs(tqqq["option_growth_overlay_enabled"], True) - self.assertEqual(tqqq["option_growth_overlay_recipe"], "tqqq_leaps_growth_v1") - self.assertEqual(tqqq["option_growth_overlay_start_usd"], 250000.0) - self.assertEqual(tqqq["option_growth_overlay_nav_budget_ratio"], 0.03) - - soxl = get_strategy_definition(SOXL_SOXX_TREND_INCOME_PROFILE).default_config - self.assertIs(soxl["option_income_overlay_enabled"], True) - self.assertEqual(soxl["option_income_overlay_recipe"], "soxx_put_credit_spread_income_v1") - self.assertEqual(soxl["option_income_overlay_start_usd"], 1000000.0) - self.assertEqual(soxl["option_income_overlay_nav_risk_ratio"], 0.01) + def test_option_overlay_defaults_are_enabled_but_live_gated(self): + live_option_profiles = ( + GLOBAL_ETF_ROTATION_PROFILE, + TQQQ_GROWTH_INCOME_PROFILE, + SOXL_SOXX_TREND_INCOME_PROFILE, + RUSSELL_TOP50_LEADER_ROTATION_PROFILE, + ) + self.assertEqual(set(OPTION_OVERLAY_DEFAULT_CONFIGS), set(live_option_profiles)) + for profile in live_option_profiles: + with self.subTest(profile=profile): + config = get_strategy_definition(profile).default_config + self.assertTrue(OPTION_OVERLAY_CONFIG_KEYS & set(config)) + self.assertIs(config["option_overlay_enabled"], True) + if config.get("option_growth_overlay_enabled"): + recipe = config["option_growth_overlay_recipe"] + self.assertEqual(OPTION_OVERLAY_RESEARCH_CANDIDATES[recipe]["status"], "research") + self.assertIs(OPTION_OVERLAY_RESEARCH_CANDIDATES[recipe]["promotion_evidence"], False) + if config.get("option_income_overlay_enabled"): + recipe = config["option_income_overlay_recipe"] + self.assertEqual(OPTION_OVERLAY_RESEARCH_CANDIDATES[recipe]["status"], "research") + self.assertIs(OPTION_OVERLAY_RESEARCH_CANDIDATES[recipe]["promotion_evidence"], False) - mega = get_strategy_definition(RUSSELL_TOP50_LEADER_ROTATION_PROFILE).default_config - self.assertIs(mega["option_growth_overlay_enabled"], True) - self.assertEqual(mega["option_growth_overlay_recipe"], "qqq_leaps_growth_v1") - self.assertEqual(mega["option_growth_overlay_start_usd"], 1000000.0) - self.assertEqual(mega["option_growth_overlay_nav_budget_ratio"], 0.03) + for profile in (NASDAQ_SP500_SMART_DCA_PROFILE, IBIT_SMART_DCA_PROFILE): + with self.subTest(profile=profile): + config = get_strategy_definition(profile).default_config + self.assertFalse(OPTION_OVERLAY_CONFIG_KEYS & set(config)) - self.assertNotIn( - "option_growth_overlay_enabled", - get_strategy_definition(NASDAQ_SP500_SMART_DCA_PROFILE).default_config, + self.assertEqual( + set(OPTION_OVERLAY_RESEARCH_CANDIDATES), + { + "tqqq_leaps_growth_v1", + "qqq_leaps_growth_v1", + "spy_leaps_growth_v1", + "soxx_put_credit_spread_income_v1", + }, ) - self.assertNotIn( - "option_growth_overlay_enabled", - get_strategy_definition(IBIT_SMART_DCA_PROFILE).default_config, + self.assertTrue( + all(not candidate["promotion_evidence"] for candidate in OPTION_OVERLAY_RESEARCH_CANDIDATES.values()) ) + def test_live_income_layer_defaults_have_validation_evidence(self): + live_income_profiles = ( + GLOBAL_ETF_ROTATION_PROFILE, + TQQQ_GROWTH_INCOME_PROFILE, + SOXL_SOXX_TREND_INCOME_PROFILE, + RUSSELL_TOP50_LEADER_ROTATION_PROFILE, + ) + self.assertEqual(set(INCOME_LAYER_LIVE_VALIDATION_EVIDENCE), set(live_income_profiles)) + for profile in live_income_profiles: + with self.subTest(profile=profile): + config = get_strategy_definition(profile).default_config + evidence = INCOME_LAYER_LIVE_VALIDATION_EVIDENCE[profile] + self.assertIs(config["income_layer_enabled"], True) + self.assertNotIn("income_threshold_usd", config) + self.assertNotIn("qqqi_income_ratio", config) + self.assertNotIn("income_layer_qqqi_weight", config) + self.assertNotIn("income_layer_spyi_weight", config) + self.assertEqual(evidence["status"], "live") + self.assertEqual(evidence["evidence_status"], "validated") + def test_dca_profiles_default_to_ordinary_dca_with_optional_smart_sizing(self): for profile in (NASDAQ_SP500_SMART_DCA_PROFILE, IBIT_SMART_DCA_PROFILE): with self.subTest(profile=profile): diff --git a/tests/test_entrypoints.py b/tests/test_entrypoints.py index 1f74c42..e926b66 100644 --- a/tests/test_entrypoints.py +++ b/tests/test_entrypoints.py @@ -8,7 +8,7 @@ from quant_platform_kit.strategy_contracts import StrategyContext from us_equity_strategies import get_platform_runtime_adapter, get_strategy_entrypoint from us_equity_strategies.catalog import get_runtime_enabled_profiles -from us_equity_strategies.entrypoints._common import OPTION_OVERLAY_CONFIG_KEYS, build_option_overlay_diagnostics +from us_equity_strategies.option_overlay import OPTION_OVERLAY_CONFIG_KEYS, build_option_overlay_diagnostics from us_equity_strategies.runtime_adapters import describe_platform_runtime_requirements from us_equity_strategies.strategies.global_etf_rotation import compute_signals_from_feature_snapshot from us_equity_strategies.strategies.tqqq_growth_income import build_rebalance_plan as tqqq_growth_build_rebalance_plan @@ -159,6 +159,52 @@ def test_option_overlay_nav_budget_ratio_is_the_only_exposed_growth_knob(self) - ) self.assertEqual(diagnostics["option_order_intents"]["intents"][0]["quantity"], 5) + def test_spy_leaps_growth_recipe_builds_index_call_intent(self) -> None: + snapshot = PortfolioSnapshot( + as_of=pd.Timestamp("2026-04-06").to_pydatetime(), + total_equity=1_000_000.0, + positions=(), + ) + + diagnostics = build_option_overlay_diagnostics( + { + "option_growth_overlay_enabled": True, + "option_growth_overlay_recipe": "spy_leaps_growth_v1", + "option_growth_overlay_start_usd": 250000.0, + }, + StrategyContext( + as_of="2026-04-06", + portfolio=snapshot, + market_data={ + "option_chains": { + "SPY": { + "contracts": ( + { + "right": "C", + "expiration": "2028-01-21", + "strike": 620.0, + "delta": 0.76, + "bid": 148.0, + "ask": 152.0, + }, + ), + }, + }, + }, + ), + ) + + self.assertIs(diagnostics["option_growth_overlay_active"], True) + self.assertEqual( + diagnostics["option_growth_overlay_recipe_detail"]["underlier"], + "SPY", + ) + option_intents = diagnostics["option_order_intents"]["intents"] + self.assertEqual(len(option_intents), 1) + self.assertEqual(option_intents[0]["action"], "buy_to_open") + self.assertEqual(option_intents[0]["underlier"], "SPY") + self.assertEqual(option_intents[0]["quantity"], 1) + def test_all_live_profiles_expose_unified_entrypoints(self) -> None: for profile in get_runtime_enabled_profiles(): entrypoint = get_strategy_entrypoint(profile) @@ -334,11 +380,8 @@ def test_tqqq_growth_income_entrypoint_maps_target_values_without_platform_layou self.assertNotIn("sell_order_symbols", decision.diagnostics) self.assertNotIn("portfolio_rows", decision.diagnostics) self.assertEqual(decision.diagnostics["threshold"], legacy_plan["threshold"]) - self.assertIs(decision.diagnostics["option_growth_overlay_enabled"], True) - self.assertEqual(decision.diagnostics["option_growth_overlay_recipe"], "tqqq_leaps_growth_v1") - self.assertEqual(decision.diagnostics["option_growth_overlay_start_usd"], 250000.0) - self.assertIs(decision.diagnostics["option_growth_overlay_active"], False) - self.assertEqual(decision.diagnostics["option_growth_overlay_skip_reason"], "below_start_usd") + self.assertNotIn("option_growth_overlay_enabled", decision.diagnostics) + self.assertNotIn("option_order_intents", decision.diagnostics) self.assertFalse(decision.diagnostics["ai_extensions"]["enabled"]) self.assertEqual(decision.diagnostics["notification_context"]["benchmark"]["symbol"], "QQQ") self.assertEqual( @@ -406,7 +449,8 @@ def test_tqqq_growth_income_defaults_to_dynamic_dual_drive_live_profile(self) -> self.assertIs(config["dual_drive_volatility_delever_taco_veto_enabled"], True) self.assertIs(config["market_regime_control_enabled"], True) self.assertEqual(config["cash_reserve_ratio"], 0.02) - self.assertEqual(config["income_threshold_usd"], 250000.0) + self.assertNotIn("income_threshold_usd", config) + self.assertNotIn("qqqi_income_ratio", config) self.assertIs(config["income_layer_enabled"], True) self.assertEqual(config["income_layer_start_usd"], 250000.0) self.assertEqual(config["income_layer_max_ratio"], 0.55) @@ -422,10 +466,7 @@ def test_tqqq_growth_income_defaults_to_dynamic_dual_drive_live_profile(self) -> {"SCHD": 0.30, "DGRO": 0.20, "SGOV": 0.40, "SPYI": 0.08, "QQQI": 0.02}, ) self.assertEqual(config["execution_cash_reserve_ratio"], 0.0) - self.assertIs(config["option_growth_overlay_enabled"], True) - self.assertEqual(config["option_growth_overlay_recipe"], "tqqq_leaps_growth_v1") - self.assertEqual(config["option_growth_overlay_start_usd"], 250000.0) - self.assertEqual(config["option_growth_overlay_nav_budget_ratio"], 0.03) + self.assertFalse(OPTION_OVERLAY_CONFIG_KEYS & set(config)) self.assertFalse(config["ai_extensions"]["enabled"]) self.assertFalse(config["ai_extensions"]["modules"]["taco_panic_rebound"]["enabled"]) self.assertFalse(config["ai_extensions"]["modules"]["crisis_regime_guard"]["enabled"]) @@ -661,15 +702,8 @@ def test_soxl_soxx_trend_income_entrypoint_maps_target_values_without_execution_ self.assertNotIn("limit_order_symbols", decision.diagnostics) self.assertNotIn("portfolio_rows", decision.diagnostics) self.assertEqual(decision.diagnostics["active_risk_asset"], legacy_plan["active_risk_asset"]) - self.assertIs(decision.diagnostics["option_income_overlay_enabled"], True) - self.assertEqual(decision.diagnostics["option_income_overlay_recipe"], "soxx_put_credit_spread_income_v1") - self.assertEqual(decision.diagnostics["option_income_overlay_start_usd"], 1000000.0) - self.assertEqual( - decision.diagnostics["option_income_overlay_recipe_detail"]["max_loss_budget_ratio"], - 0.01, - ) - self.assertIs(decision.diagnostics["option_income_overlay_active"], False) - self.assertEqual(decision.diagnostics["option_income_overlay_skip_reason"], "below_start_usd") + self.assertNotIn("option_income_overlay_enabled", decision.diagnostics) + self.assertNotIn("option_order_intents", decision.diagnostics) self.assertEqual( decision.diagnostics["notification_context"]["status"]["code"], legacy_plan["notification_context"]["status"]["code"], diff --git a/tests/test_option_overlay.py b/tests/test_option_overlay.py new file mode 100644 index 0000000..d889540 --- /dev/null +++ b/tests/test_option_overlay.py @@ -0,0 +1,290 @@ +from __future__ import annotations + +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import patch + +from quant_platform_kit.strategy_contracts import StrategyContext +from us_equity_strategies.option_overlay import ( + OPTION_OVERLAY_CONFIG_KEYS, + OPTION_OVERLAY_DEFAULT_CONFIGS, + OPTION_OVERLAY_RECIPE_DETAILS, + OPTION_OVERLAY_RESEARCH_CANDIDATES, + build_option_overlay_diagnostics, + option_overlay_default_config, +) + + +REPO_ROOT = Path(__file__).resolve().parents[1] + + +class OptionOverlayArchitectureTest(unittest.TestCase): + def test_option_overlay_config_keys_stay_out_of_income_layer_namespace(self) -> None: + for key in OPTION_OVERLAY_CONFIG_KEYS: + self.assertFalse(key.startswith("income_layer_")) + + def test_option_overlay_recipes_are_research_candidates_until_promoted(self) -> None: + self.assertEqual(set(OPTION_OVERLAY_RESEARCH_CANDIDATES), set(OPTION_OVERLAY_RECIPE_DETAILS)) + for candidate in OPTION_OVERLAY_RESEARCH_CANDIDATES.values(): + self.assertEqual(candidate["status"], "research") + self.assertIs(candidate["promotion_evidence"], False) + + def test_default_option_overlay_configs_are_enabled_but_scoped(self) -> None: + self.assertEqual( + set(OPTION_OVERLAY_DEFAULT_CONFIGS), + { + "global_etf_rotation", + "tqqq_growth_income", + "soxl_soxx_trend_income", + "russell_top50_leader_rotation", + }, + ) + tqqq = option_overlay_default_config("tqqq_growth_income") + self.assertIs(tqqq["option_overlay_enabled"], True) + self.assertIs(tqqq["option_growth_overlay_enabled"], True) + self.assertEqual(tqqq["option_growth_overlay_recipe"], "tqqq_leaps_growth_v1") + + cloned = option_overlay_default_config("tqqq_growth_income") + cloned["option_growth_overlay_nav_budget_ratio"] = 0.01 + self.assertEqual( + option_overlay_default_config("tqqq_growth_income")["option_growth_overlay_nav_budget_ratio"], + 0.03, + ) + self.assertEqual(option_overlay_default_config("nasdaq_sp500_smart_dca"), {}) + + def test_option_overlay_module_has_no_income_layer_dependency(self) -> None: + source = (REPO_ROOT / "src/us_equity_strategies/option_overlay.py").read_text(encoding="utf-8") + self.assertNotIn("income_layer", source) + + def test_entrypoint_common_reexports_option_overlay_contract_by_import(self) -> None: + source = (REPO_ROOT / "src/us_equity_strategies/entrypoints/_common.py").read_text(encoding="utf-8") + self.assertIn("from us_equity_strategies.option_overlay import", source) + self.assertIn("OPTION_OVERLAY_CONFIG_KEYS", source) + self.assertIn("build_option_overlay_diagnostics", source) + + def test_master_option_overlay_switch_disables_all_option_intents(self) -> None: + snapshot = SimpleNamespace(total_equity=2_000_000.0, positions=(), metadata={}) + + diagnostics = build_option_overlay_diagnostics( + { + "option_overlay_enabled": False, + "option_growth_overlay_enabled": True, + "option_growth_overlay_recipe": "spy_leaps_growth_v1", + "option_growth_overlay_start_usd": 250000.0, + }, + StrategyContext( + as_of="2026-04-06", + portfolio=snapshot, + market_data={ + "option_chains": { + "SPY": { + "contracts": ( + { + "right": "C", + "expiration": "2028-01-21", + "strike": 620.0, + "delta": 0.76, + "bid": 148.0, + "ask": 152.0, + }, + ), + }, + }, + }, + ), + ) + + self.assertIs(diagnostics["option_overlay_enabled"], False) + self.assertIs(diagnostics["option_growth_overlay_enabled"], True) + self.assertIs(diagnostics["option_growth_overlay_active"], False) + self.assertEqual(diagnostics["option_growth_overlay_skip_reason"], "option_overlay_disabled") + self.assertNotIn("option_order_intents", diagnostics) + + def test_research_recipe_blocks_live_intents_even_when_default_enabled(self) -> None: + snapshot = SimpleNamespace(total_equity=2_000_000.0, positions=(), metadata={}) + + diagnostics = build_option_overlay_diagnostics( + option_overlay_default_config("global_etf_rotation"), + StrategyContext( + as_of="2026-04-06", + portfolio=snapshot, + market_data={ + "underlier_indicators": { + "SPY": { + "sma200_pass": True, + "momentum_63d": 0.04, + }, + }, + "option_chains": { + "SPY": { + "contracts": ( + { + "right": "C", + "expiration": "2028-01-21", + "strike": 620.0, + "delta": 0.76, + "bid": 148.0, + "ask": 152.0, + }, + ), + }, + }, + }, + ), + base_diagnostics={"regime": "risk_on"}, + ) + + self.assertIs(diagnostics["option_overlay_enabled"], True) + self.assertIs(diagnostics["option_growth_overlay_enabled"], True) + self.assertIs(diagnostics["option_growth_overlay_live_allowed"], False) + self.assertEqual(diagnostics["option_growth_overlay_promotion_status"], "research") + self.assertIs(diagnostics["option_growth_overlay_active"], False) + self.assertEqual(diagnostics["option_growth_overlay_skip_reason"], "research_only_recipe") + self.assertNotIn("option_order_intents", diagnostics) + + def test_index_leaps_growth_recipe_respects_risk_off_regime_gate(self) -> None: + snapshot = SimpleNamespace(total_equity=2_000_000.0, positions=(), metadata={}) + + with patch.dict( + OPTION_OVERLAY_RESEARCH_CANDIDATES, + {"spy_leaps_growth_v1": {"status": "live", "promotion_evidence": True, "reason": "test"}}, + ): + diagnostics = build_option_overlay_diagnostics( + { + "option_growth_overlay_enabled": True, + "option_growth_overlay_recipe": "spy_leaps_growth_v1", + "option_growth_overlay_start_usd": 250000.0, + }, + StrategyContext( + as_of="2026-04-06", + portfolio=snapshot, + market_data={ + "option_chains": { + "SPY": { + "contracts": ( + { + "right": "C", + "expiration": "2028-01-21", + "strike": 620.0, + "delta": 0.76, + "bid": 148.0, + "ask": 152.0, + }, + ), + }, + }, + }, + ), + base_diagnostics={"regime": "risk_off"}, + ) + + self.assertIs(diagnostics["option_growth_overlay_active"], True) + self.assertEqual(diagnostics["option_order_intent_count"], 0) + self.assertEqual( + diagnostics["option_order_intents"]["skipped"], + ({"recipe": "spy_leaps_growth_v1", "underlier": "SPY", "reason": "entry_gate_not_met"},), + ) + + def test_index_leaps_growth_recipe_requires_positive_underlier_trend_when_indicators_exist(self) -> None: + snapshot = SimpleNamespace(total_equity=2_000_000.0, positions=(), metadata={}) + + with patch.dict( + OPTION_OVERLAY_RESEARCH_CANDIDATES, + {"qqq_leaps_growth_v1": {"status": "live", "promotion_evidence": True, "reason": "test"}}, + ): + diagnostics = build_option_overlay_diagnostics( + { + "option_overlay_enabled": True, + "option_growth_overlay_enabled": True, + "option_growth_overlay_recipe": "qqq_leaps_growth_v1", + "option_growth_overlay_start_usd": 250000.0, + }, + StrategyContext( + as_of="2026-04-06", + portfolio=snapshot, + market_data={ + "underlier_indicators": { + "QQQ": { + "above_200dma": True, + "momentum_63d": -0.02, + }, + }, + "option_chains": { + "QQQ": { + "contracts": ( + { + "right": "C", + "expiration": "2028-01-21", + "strike": 500.0, + "delta": 0.75, + "bid": 120.0, + "ask": 124.0, + }, + ), + }, + }, + }, + ), + base_diagnostics={"regime": "risk_on"}, + ) + + self.assertIs(diagnostics["option_growth_overlay_active"], True) + self.assertEqual(diagnostics["option_order_intent_count"], 0) + self.assertEqual( + diagnostics["option_order_intents"]["skipped"], + ({"recipe": "qqq_leaps_growth_v1", "underlier": "QQQ", "reason": "entry_gate_not_met"},), + ) + + def test_index_leaps_growth_recipe_allows_entry_when_regime_and_indicators_pass(self) -> None: + snapshot = SimpleNamespace(total_equity=2_000_000.0, positions=(), metadata={}) + + with patch.dict( + OPTION_OVERLAY_RESEARCH_CANDIDATES, + {"spy_leaps_growth_v1": {"status": "live", "promotion_evidence": True, "reason": "test"}}, + ): + diagnostics = build_option_overlay_diagnostics( + { + "option_overlay_enabled": True, + "option_growth_overlay_enabled": True, + "option_growth_overlay_recipe": "spy_leaps_growth_v1", + "option_growth_overlay_start_usd": 250000.0, + "option_growth_overlay_nav_budget_ratio": 0.015, + }, + StrategyContext( + as_of="2026-04-06", + portfolio=snapshot, + market_data={ + "underlier_indicators": { + "SPY": { + "sma200_pass": True, + "momentum_63d": 0.04, + }, + }, + "option_chains": { + "SPY": { + "contracts": ( + { + "right": "C", + "expiration": "2028-01-21", + "strike": 620.0, + "delta": 0.76, + "bid": 148.0, + "ask": 152.0, + }, + ), + }, + }, + }, + ), + base_diagnostics={"regime": "risk_on"}, + ) + + self.assertEqual(diagnostics["option_order_intent_count"], 1) + intent = diagnostics["option_order_intents"]["intents"][0] + self.assertEqual(intent["underlier"], "SPY") + self.assertEqual(intent["action"], "buy_to_open") + + +if __name__ == "__main__": + unittest.main() From 59fc3956057ca58d6f81b4f8ffac71f63fe8c50f Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 23 Jun 2026 05:10:41 +0800 Subject: [PATCH 2/2] Update entrypoint tests for live-gated option overlays --- tests/test_entrypoints.py | 188 ++++++++++++++++++++++---------------- 1 file changed, 108 insertions(+), 80 deletions(-) diff --git a/tests/test_entrypoints.py b/tests/test_entrypoints.py index e926b66..f675f33 100644 --- a/tests/test_entrypoints.py +++ b/tests/test_entrypoints.py @@ -1,6 +1,7 @@ from __future__ import annotations import unittest +from unittest.mock import patch import pandas as pd @@ -8,7 +9,11 @@ from quant_platform_kit.strategy_contracts import StrategyContext from us_equity_strategies import get_platform_runtime_adapter, get_strategy_entrypoint from us_equity_strategies.catalog import get_runtime_enabled_profiles -from us_equity_strategies.option_overlay import OPTION_OVERLAY_CONFIG_KEYS, build_option_overlay_diagnostics +from us_equity_strategies.option_overlay import ( + OPTION_OVERLAY_CONFIG_KEYS, + OPTION_OVERLAY_RESEARCH_CANDIDATES, + build_option_overlay_diagnostics, +) from us_equity_strategies.runtime_adapters import describe_platform_runtime_requirements from us_equity_strategies.strategies.global_etf_rotation import compute_signals_from_feature_snapshot from us_equity_strategies.strategies.tqqq_growth_income import build_rebalance_plan as tqqq_growth_build_rebalance_plan @@ -77,33 +82,37 @@ def test_option_overlay_diagnostics_respect_start_threshold(self) -> None: positions=(), ) - diagnostics = build_option_overlay_diagnostics( - { - "option_growth_overlay_enabled": True, - "option_growth_overlay_recipe": "tqqq_leaps_growth_v1", - "option_growth_overlay_start_usd": 250000.0, - }, - StrategyContext( - as_of="2026-04-06", - portfolio=snapshot, - market_data={ - "option_chains": { - "TQQQ": { - "contracts": ( - { - "right": "C", - "expiration": "2028-01-21", - "strike": 70.0, - "delta": 0.74, - "bid": 29.0, - "ask": 31.0, - }, - ), + with patch.dict( + OPTION_OVERLAY_RESEARCH_CANDIDATES, + {"tqqq_leaps_growth_v1": {"status": "live", "promotion_evidence": True, "reason": "test"}}, + ): + diagnostics = build_option_overlay_diagnostics( + { + "option_growth_overlay_enabled": True, + "option_growth_overlay_recipe": "tqqq_leaps_growth_v1", + "option_growth_overlay_start_usd": 250000.0, + }, + StrategyContext( + as_of="2026-04-06", + portfolio=snapshot, + market_data={ + "option_chains": { + "TQQQ": { + "contracts": ( + { + "right": "C", + "expiration": "2028-01-21", + "strike": 70.0, + "delta": 0.74, + "bid": 29.0, + "ask": 31.0, + }, + ), + }, }, }, - }, - ), - ) + ), + ) self.assertIs(diagnostics["option_growth_overlay_active"], True) self.assertNotIn("option_growth_overlay_skip_reason", diagnostics) @@ -124,34 +133,38 @@ def test_option_overlay_nav_budget_ratio_is_the_only_exposed_growth_knob(self) - positions=(), ) - diagnostics = build_option_overlay_diagnostics( - { - "option_growth_overlay_enabled": True, - "option_growth_overlay_recipe": "tqqq_leaps_growth_v1", - "option_growth_overlay_start_usd": 250000.0, - "option_growth_overlay_nav_budget_ratio": 0.06, - }, - StrategyContext( - as_of="2026-04-06", - portfolio=snapshot, - market_data={ - "option_chains": { - "TQQQ": { - "contracts": ( - { - "right": "C", - "expiration": "2028-01-21", - "strike": 70.0, - "delta": 0.74, - "bid": 29.0, - "ask": 31.0, - }, - ), + with patch.dict( + OPTION_OVERLAY_RESEARCH_CANDIDATES, + {"tqqq_leaps_growth_v1": {"status": "live", "promotion_evidence": True, "reason": "test"}}, + ): + diagnostics = build_option_overlay_diagnostics( + { + "option_growth_overlay_enabled": True, + "option_growth_overlay_recipe": "tqqq_leaps_growth_v1", + "option_growth_overlay_start_usd": 250000.0, + "option_growth_overlay_nav_budget_ratio": 0.06, + }, + StrategyContext( + as_of="2026-04-06", + portfolio=snapshot, + market_data={ + "option_chains": { + "TQQQ": { + "contracts": ( + { + "right": "C", + "expiration": "2028-01-21", + "strike": 70.0, + "delta": 0.74, + "bid": 29.0, + "ask": 31.0, + }, + ), + }, }, }, - }, - ), - ) + ), + ) self.assertEqual( diagnostics["option_growth_overlay_recipe_detail"]["premium_budget_ratio"], @@ -166,33 +179,37 @@ def test_spy_leaps_growth_recipe_builds_index_call_intent(self) -> None: positions=(), ) - diagnostics = build_option_overlay_diagnostics( - { - "option_growth_overlay_enabled": True, - "option_growth_overlay_recipe": "spy_leaps_growth_v1", - "option_growth_overlay_start_usd": 250000.0, - }, - StrategyContext( - as_of="2026-04-06", - portfolio=snapshot, - market_data={ - "option_chains": { - "SPY": { - "contracts": ( - { - "right": "C", - "expiration": "2028-01-21", - "strike": 620.0, - "delta": 0.76, - "bid": 148.0, - "ask": 152.0, - }, - ), + with patch.dict( + OPTION_OVERLAY_RESEARCH_CANDIDATES, + {"spy_leaps_growth_v1": {"status": "live", "promotion_evidence": True, "reason": "test"}}, + ): + diagnostics = build_option_overlay_diagnostics( + { + "option_growth_overlay_enabled": True, + "option_growth_overlay_recipe": "spy_leaps_growth_v1", + "option_growth_overlay_start_usd": 250000.0, + }, + StrategyContext( + as_of="2026-04-06", + portfolio=snapshot, + market_data={ + "option_chains": { + "SPY": { + "contracts": ( + { + "right": "C", + "expiration": "2028-01-21", + "strike": 620.0, + "delta": 0.76, + "bid": 148.0, + "ask": 152.0, + }, + ), + }, }, }, - }, - ), - ) + ), + ) self.assertIs(diagnostics["option_growth_overlay_active"], True) self.assertEqual( @@ -380,7 +397,9 @@ def test_tqqq_growth_income_entrypoint_maps_target_values_without_platform_layou self.assertNotIn("sell_order_symbols", decision.diagnostics) self.assertNotIn("portfolio_rows", decision.diagnostics) self.assertEqual(decision.diagnostics["threshold"], legacy_plan["threshold"]) - self.assertNotIn("option_growth_overlay_enabled", decision.diagnostics) + self.assertIs(decision.diagnostics["option_growth_overlay_enabled"], True) + self.assertIs(decision.diagnostics["option_growth_overlay_live_allowed"], False) + self.assertEqual(decision.diagnostics["option_growth_overlay_skip_reason"], "research_only_recipe") self.assertNotIn("option_order_intents", decision.diagnostics) self.assertFalse(decision.diagnostics["ai_extensions"]["enabled"]) self.assertEqual(decision.diagnostics["notification_context"]["benchmark"]["symbol"], "QQQ") @@ -466,7 +485,14 @@ def test_tqqq_growth_income_defaults_to_dynamic_dual_drive_live_profile(self) -> {"SCHD": 0.30, "DGRO": 0.20, "SGOV": 0.40, "SPYI": 0.08, "QQQI": 0.02}, ) self.assertEqual(config["execution_cash_reserve_ratio"], 0.0) - self.assertFalse(OPTION_OVERLAY_CONFIG_KEYS & set(config)) + self.assertIs(config["option_overlay_enabled"], True) + self.assertIs(config["option_growth_overlay_enabled"], True) + self.assertEqual(config["option_growth_overlay_recipe"], "tqqq_leaps_growth_v1") + self.assertEqual(config["option_growth_overlay_start_usd"], 250000.0) + self.assertEqual(config["option_growth_overlay_nav_budget_ratio"], 0.03) + self.assertIs(config["option_income_overlay_enabled"], False) + self.assertEqual(OPTION_OVERLAY_RESEARCH_CANDIDATES["tqqq_leaps_growth_v1"]["status"], "research") + self.assertIs(OPTION_OVERLAY_RESEARCH_CANDIDATES["tqqq_leaps_growth_v1"]["promotion_evidence"], False) self.assertFalse(config["ai_extensions"]["enabled"]) self.assertFalse(config["ai_extensions"]["modules"]["taco_panic_rebound"]["enabled"]) self.assertFalse(config["ai_extensions"]["modules"]["crisis_regime_guard"]["enabled"]) @@ -702,7 +728,9 @@ def test_soxl_soxx_trend_income_entrypoint_maps_target_values_without_execution_ self.assertNotIn("limit_order_symbols", decision.diagnostics) self.assertNotIn("portfolio_rows", decision.diagnostics) self.assertEqual(decision.diagnostics["active_risk_asset"], legacy_plan["active_risk_asset"]) - self.assertNotIn("option_income_overlay_enabled", decision.diagnostics) + self.assertIs(decision.diagnostics["option_income_overlay_enabled"], True) + self.assertIs(decision.diagnostics["option_income_overlay_live_allowed"], False) + self.assertEqual(decision.diagnostics["option_income_overlay_skip_reason"], "research_only_recipe") self.assertNotIn("option_order_intents", decision.diagnostics) self.assertEqual( decision.diagnostics["notification_context"]["status"]["code"],