From fb778d0b0ddc3b9d0f321adb2946129c31fc92f3 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 5 Apr 2026 04:51:55 +0800 Subject: [PATCH 1/3] add cash buffer branch default --- README.md | 62 ++ ...ate_cash_buffer_branch_feature_snapshot.py | 124 +++ src/us_equity_strategies/catalog.py | 12 + .../snapshots/__init__.py | 3 + .../snapshots/cash_buffer_branch_default.py | 328 ++++++++ .../strategies/cash_buffer_branch_default.py | 736 ++++++++++++++++++ tests/test_cash_buffer_branch_default.py | 166 ++++ ...est_cash_buffer_branch_feature_snapshot.py | 97 +++ tests/test_catalog.py | 13 + 9 files changed, 1541 insertions(+) create mode 100644 scripts/generate_cash_buffer_branch_feature_snapshot.py create mode 100644 src/us_equity_strategies/snapshots/__init__.py create mode 100644 src/us_equity_strategies/snapshots/cash_buffer_branch_default.py create mode 100644 src/us_equity_strategies/strategies/cash_buffer_branch_default.py create mode 100644 tests/test_cash_buffer_branch_default.py create mode 100644 tests/test_cash_buffer_branch_feature_snapshot.py diff --git a/README.md b/README.md index 1995acfb..d739990c 100644 --- a/README.md +++ b/README.md @@ -16,6 +16,7 @@ This repository is the strategy layer: it owns pure signal, allocation, and targ | Profile | Downstream runtime today | Core idea | | --- | --- | --- | | `global_etf_rotation` | `InteractiveBrokersPlatform` | Quarterly top-2 global ETF rotation with a daily canary defense | +| `cash_buffer_branch_default` | `InteractiveBrokersPlatform` | Tech-heavy monthly stock selection with an explicit 20% BOXX/cash buffer in risk-on and a QQQ+breadth defense ladder | | `hybrid_growth_income` | `CharlesSchwabPlatform` | QQQ-driven TQQQ attack layer plus SPYI / QQQI income layer and BOXX defense | | `semiconductor_rotation_income` | `LongBridgePlatform` | SOXL / SOXX trend switch with BOXX parking and an additive income sleeve | @@ -50,6 +51,26 @@ These strategies are consumed by platform repositories through `QuantPlatformKit - Compared with a pure tech or leveraged-Nasdaq approach, this profile is meant to be steadier. - It still allows `VOO`, `XLK`, and `SMH` to win their way into the rotation instead of hard-coding them out. +### cash_buffer_branch_default + +**Objective** +- Provide a research-only but runtime-loadable stock branch for concentrated tech leaders bought on controlled pullbacks. +- Keep the branch geometry honest: in `risk_on`, the profile explicitly targets `80%` stock exposure and parks the rest in `BOXX` / cash instead of relying on accidental underinvestment. + +**Current default shape** +- Universe: large-cap US tech / communication names from the shared snapshot task +- Benchmark: `QQQ` +- Safe haven: `BOXX` +- Position count: `8` +- Single-name cap: `10%` +- Sector cap: `40%` +- Default exposures: `80% / 60% / 0%` + +**Current runtime contract** +- Consumes a precomputed feature snapshot plus a sidecar manifest +- Expects the canonical config name `cash_buffer_branch_default` +- Designed for monthly execution windows; non-window runs should no-op + ### hybrid_growth_income **Objective** @@ -150,6 +171,7 @@ These strategies are consumed by platform repositories through `QuantPlatformKit | 策略档位 | 当前下游运行仓库 | 核心思路 | | --- | --- | --- | | `global_etf_rotation` | `InteractiveBrokersPlatform` | 22 只全球 ETF 的季度 Top 2 轮动，带每日 canary 防守 | +| `cash_buffer_branch_default` | `InteractiveBrokersPlatform` | 偏科技个股的月频受控回调分支，`risk_on` 明确只上 `80%` 股票，其余停在 `BOXX` / 现金 | | `hybrid_growth_income` | `CharlesSchwabPlatform` | 由 QQQ 驱动的 TQQQ 攻击层，加上 SPYI / QQQI 收入层和 BOXX 防守层 | | `semiconductor_rotation_income` | `LongBridgePlatform` | SOXL / SOXX 趋势切换，剩余资金停在 BOXX，并叠加收入层 | @@ -184,6 +206,46 @@ These strategies are consumed by platform repositories through `QuantPlatformKit - 相比纯科技或者杠杆纳指路线，这个档位更稳。 - 但它仍然允许 `VOO`、`XLK`、`SMH` 靠表现进入组合，而不是事先把它们排除。 +### cash_buffer_branch_default + +**Objective** +- Provide a research-only but runtime-loadable stock branch for concentrated tech leaders bought on controlled pullbacks. +- Keep the branch geometry honest: in `risk_on`, the profile explicitly targets `80%` stock exposure and parks the rest in `BOXX` / cash instead of relying on accidental underinvestment. + +**Current default shape** +- Universe: large-cap US tech / communication names from the shared snapshot task +- Benchmark: `QQQ` +- Safe haven: `BOXX` +- Position count: `8` +- Single-name cap: `10%` +- Sector cap: `40%` +- Default exposures: `80% / 60% / 0%` + +**Current runtime contract** +- Consumes a precomputed feature snapshot plus a sidecar manifest +- Expects the canonical config name `cash_buffer_branch_default` +- Designed for monthly execution windows; non-window runs should no-op + +### cash_buffer_branch_default + +**策略目标** +- 提供一条研究优先、但已经能被下游 runtime 正式加载的个股分支。 +- 核心是做偏科技龙头的受控回调买入，同时把 `risk_on` 的 `80%` 股票暴露显式写进规格里，不再靠隐式留仓位。 + +**当前默认规格** +- 股票池：共享 snapshot 任务里可交易的大盘科技 / 通信股票 +- 基准：`QQQ` +- 防守腿：`BOXX` +- 持仓数：`8` +- 单票上限：`10%` +- 行业上限：`40%` +- 默认暴露：`80% / 60% / 0%` + +**运行约定** +- 运行时消费预先生成好的 feature snapshot 和 sidecar manifest +- canonical config 名必须是 `cash_buffer_branch_default` +- 设计上保持月频执行，非执行窗口应显式 no-op + ### hybrid_growth_income **策略目标** diff --git a/scripts/generate_cash_buffer_branch_feature_snapshot.py b/scripts/generate_cash_buffer_branch_feature_snapshot.py new file mode 100644 index 00000000..b30258aa --- /dev/null +++ b/scripts/generate_cash_buffer_branch_feature_snapshot.py @@ -0,0 +1,124 @@ +from __future__ import annotations + +import argparse +import hashlib +import json +from datetime import datetime, timezone +from pathlib import Path + +from us_equity_strategies.snapshots.cash_buffer_branch_default import ( + build_feature_snapshot, + read_table, + write_table, +) +from us_equity_strategies.strategies.cash_buffer_branch_default import ( + PROFILE_NAME, + SNAPSHOT_CONTRACT_VERSION, +) + + +def _sha256_file(path: Path) -> str: + hasher = hashlib.sha256() + with path.open("rb") as fh: + while True: + chunk = fh.read(1024 * 1024) + if not chunk: + break + hasher.update(chunk) + return hasher.hexdigest() + + +def _default_config_path() -> Path | None: + sibling = ( + Path(__file__).resolve().parents[2] + / "InteractiveBrokersPlatform" + / "research" + / "configs" + / "growth_pullback_cash_buffer_branch_default.json" + ) + return sibling if sibling.exists() else None + + +def write_snapshot_manifest( + *, + snapshot_path: Path, + snapshot, + config_path: Path | None, + manifest_path: Path | None = None, +) -> Path: + resolved_manifest = manifest_path or Path(f"{snapshot_path}.manifest.json") + if config_path is None or not config_path.exists(): + raise FileNotFoundError( + f"cash_buffer_branch_default snapshot manifest requires a valid config_path, got: {config_path}" + ) + config_payload = json.loads(config_path.read_text(encoding="utf-8")) + config_sha256 = _sha256_file(config_path) + payload = { + "manifest_type": "feature_snapshot", + "contract_version": SNAPSHOT_CONTRACT_VERSION, + "strategy_profile": PROFILE_NAME, + "config_name": str(config_payload.get("name") or PROFILE_NAME), + "config_path": str(config_path) if config_path is not None else None, + "config_sha256": config_sha256, + "snapshot_path": str(snapshot_path), + "snapshot_sha256": _sha256_file(snapshot_path), + "snapshot_as_of": str(snapshot["as_of"].max()), + "row_count": int(len(snapshot)), + "generated_at": datetime.now(timezone.utc).isoformat(), + } + resolved_manifest.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") + return resolved_manifest + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="Generate a cash_buffer_branch_default feature snapshot.", + ) + parser.add_argument("--prices", required=True, help="Input price history file (.csv/.json/.jsonl/.parquet)") + parser.add_argument("--universe", required=True, help="Input universe file (.csv/.json/.jsonl/.parquet)") + parser.add_argument("--output", required=True, help="Output feature snapshot path") + parser.add_argument("--manifest-output", default=None, help="Optional output path for sidecar manifest JSON") + parser.add_argument( + "--config-path", + default=str(_default_config_path()) if _default_config_path() is not None else None, + help="Canonical strategy config path used to populate manifest metadata", + ) + parser.add_argument("--as-of", dest="as_of_date", required=True, help="Snapshot date") + parser.add_argument("--benchmark-symbol", default="QQQ") + parser.add_argument("--safe-haven", default="BOXX") + parser.add_argument("--min-price-usd", type=float, default=10.0) + parser.add_argument("--min-adv20-usd", type=float, default=50_000_000.0) + parser.add_argument("--min-history-days", type=int, default=252) + return parser + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + + price_history = read_table(args.prices) + universe_snapshot = read_table(args.universe) + snapshot = build_feature_snapshot( + price_history, + universe_snapshot, + as_of_date=args.as_of_date, + benchmark_symbol=args.benchmark_symbol, + safe_haven=args.safe_haven, + min_price_usd=args.min_price_usd, + min_adv20_usd=args.min_adv20_usd, + min_history_days=args.min_history_days, + ) + write_table(snapshot, args.output) + manifest_path = write_snapshot_manifest( + snapshot_path=Path(args.output), + snapshot=snapshot, + config_path=Path(args.config_path) if args.config_path else None, + manifest_path=Path(args.manifest_output) if args.manifest_output else None, + ) + print(f"wrote {len(snapshot)} rows -> {Path(args.output)}") + print(f"wrote manifest -> {manifest_path}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/us_equity_strategies/catalog.py b/src/us_equity_strategies/catalog.py index 363f520d..22849a01 100644 --- a/src/us_equity_strategies/catalog.py +++ b/src/us_equity_strategies/catalog.py @@ -9,6 +9,7 @@ GLOBAL_ETF_ROTATION_PROFILE = "global_etf_rotation" HYBRID_GROWTH_INCOME_PROFILE = "hybrid_growth_income" SEMICONDUCTOR_ROTATION_INCOME_PROFILE = "semiconductor_rotation_income" +CASH_BUFFER_BRANCH_DEFAULT_PROFILE = "cash_buffer_branch_default" STRATEGY_DEFINITIONS: dict[str, StrategyDefinition] = { GLOBAL_ETF_ROTATION_PROFILE: StrategyDefinition( @@ -44,6 +45,17 @@ ), ), ), + CASH_BUFFER_BRANCH_DEFAULT_PROFILE: StrategyDefinition( + profile=CASH_BUFFER_BRANCH_DEFAULT_PROFILE, + domain=US_EQUITY_DOMAIN, + supported_platforms=frozenset({"ibkr"}), + components=( + StrategyComponentDefinition( + name="signal_logic", + module_path="us_equity_strategies.strategies.cash_buffer_branch_default", + ), + ), + ), } diff --git a/src/us_equity_strategies/snapshots/__init__.py b/src/us_equity_strategies/snapshots/__init__.py new file mode 100644 index 00000000..95c63ca8 --- /dev/null +++ b/src/us_equity_strategies/snapshots/__init__.py @@ -0,0 +1,3 @@ +"""Snapshot builders for research and runtime feature files.""" + +__all__ = [] diff --git a/src/us_equity_strategies/snapshots/cash_buffer_branch_default.py b/src/us_equity_strategies/snapshots/cash_buffer_branch_default.py new file mode 100644 index 00000000..58c1ea4b --- /dev/null +++ b/src/us_equity_strategies/snapshots/cash_buffer_branch_default.py @@ -0,0 +1,328 @@ +from __future__ import annotations + +import math +from collections.abc import Mapping +from pathlib import Path + +import numpy as np +import pandas as pd + +from us_equity_strategies.strategies.cash_buffer_branch_default import ( + BENCHMARK_SYMBOL, + DEFAULT_MIN_ADV20_USD, + DEFAULT_SECTOR_WHITELIST, + PROFILE_NAME, + SAFE_HAVEN, +) + +PRICE_HISTORY_REQUIRED_COLUMNS = frozenset({"symbol", "as_of", "close", "volume"}) +UNIVERSE_REQUIRED_COLUMNS = frozenset({"symbol", "sector"}) + + +def resolve_active_universe(universe_snapshot: pd.DataFrame, as_of_date) -> pd.DataFrame: + as_of = pd.Timestamp(as_of_date).tz_localize(None).normalize() + frame = universe_snapshot.copy() + + if "start_date" in frame.columns: + frame = frame.loc[frame["start_date"].isna() | (frame["start_date"] <= as_of)] + if "end_date" in frame.columns: + frame = frame.loc[frame["end_date"].isna() | (frame["end_date"] >= as_of)] + + return frame.loc[:, ["symbol", "sector"]].drop_duplicates(subset=["symbol"], keep="last").reset_index(drop=True) + + +def read_table(path: str | Path) -> pd.DataFrame: + raw_path = str(path or "").strip() + if not raw_path: + raise EnvironmentError("path is required") + table_path = Path(raw_path) + if not table_path.exists(): + raise FileNotFoundError(f"file not found: {table_path}") + + suffix = table_path.suffix.lower() + if suffix == ".csv": + return pd.read_csv(table_path) + if suffix in {".json", ".jsonl"}: + return pd.read_json(table_path, orient="records", lines=suffix == ".jsonl") + if suffix == ".parquet": + return pd.read_parquet(table_path) + raise ValueError("Unsupported table format; expected .csv, .json, .jsonl, or .parquet") + + +def write_table(frame: pd.DataFrame, path: str | Path) -> None: + raw_path = str(path or "").strip() + if not raw_path: + raise EnvironmentError("path is required") + table_path = Path(raw_path) + table_path.parent.mkdir(parents=True, exist_ok=True) + + suffix = table_path.suffix.lower() + if suffix == ".csv": + frame.to_csv(table_path, index=False) + return + if suffix == ".json": + frame.to_json(table_path, orient="records", indent=2, date_format="iso") + return + if suffix == ".jsonl": + frame.to_json(table_path, orient="records", lines=True, date_format="iso") + return + if suffix == ".parquet": + frame.to_parquet(table_path, index=False) + return + raise ValueError("Unsupported table format; expected .csv, .json, .jsonl, or .parquet") + + +FEATURE_SNAPSHOT_COLUMNS = ( + "as_of", + "symbol", + "sector", + "close", + "volume", + "adv20_usd", + "history_days", + "mom_6_1", + "mom_12_1", + "sma20_gap", + "sma50_gap", + "sma200_gap", + "ma50_over_ma200", + "vol_63", + "maxdd_126", + "breakout_252", + "dist_63_high", + "dist_126_high", + "rebound_20", + "base_eligible", +) + + +def _require_columns(frame: pd.DataFrame, required: frozenset[str], *, name: str) -> None: + missing = required - set(frame.columns) + if missing: + missing_text = ", ".join(sorted(missing)) + raise ValueError(f"{name} missing required columns: {missing_text}") + + +def _normalize_symbol_series(values: pd.Series) -> pd.Series: + return values.astype(str).str.upper().str.strip() + + +def _normalize_date(value) -> pd.Timestamp: + timestamp = pd.Timestamp(value) + if pd.isna(timestamp): + return pd.NaT + if timestamp.tzinfo is not None: + timestamp = timestamp.tz_convert(None) + return timestamp.normalize() + + +def _normalize_price_groups( + price_history, + *, + as_of: pd.Timestamp, +) -> tuple[dict[str, pd.DataFrame], pd.DataFrame]: + if isinstance(price_history, Mapping): + price_groups: dict[str, pd.DataFrame] = {} + empty_history = pd.DataFrame(columns=["symbol", "as_of", "close", "volume"]) + for raw_symbol, raw_history in price_history.items(): + history = pd.DataFrame(raw_history).copy() + if history.empty: + continue + _require_columns(history, PRICE_HISTORY_REQUIRED_COLUMNS, name=f"price_history[{raw_symbol!r}]") + history["symbol"] = _normalize_symbol_series(history["symbol"]) + history["as_of"] = pd.to_datetime(history["as_of"], utc=False).map(_normalize_date) + history["close"] = pd.to_numeric(history["close"], errors="coerce") + history["volume"] = pd.to_numeric(history["volume"], errors="coerce") + history = history.dropna(subset=["symbol", "as_of", "close"]) + history = history.loc[history["as_of"] <= as_of].sort_values("as_of").reset_index(drop=True) + if history.empty: + continue + price_groups[str(raw_symbol).strip().upper()] = history + if empty_history.empty: + empty_history = history.iloc[0:0].copy() + return price_groups, empty_history + + prices = pd.DataFrame(price_history).copy() + if prices.empty: + raise ValueError("price_history must contain at least one row") + _require_columns(prices, PRICE_HISTORY_REQUIRED_COLUMNS, name="price_history") + + prices["symbol"] = _normalize_symbol_series(prices["symbol"]) + prices["as_of"] = pd.to_datetime(prices["as_of"], utc=False).map(_normalize_date) + prices["close"] = pd.to_numeric(prices["close"], errors="coerce") + prices["volume"] = pd.to_numeric(prices["volume"], errors="coerce") + prices = prices.dropna(subset=["symbol", "as_of", "close"]) + prices = prices.loc[prices["as_of"] <= as_of].copy() + price_groups = { + symbol: group.sort_values("as_of").reset_index(drop=True) + for symbol, group in prices.groupby("symbol", sort=False) + } + return price_groups, prices.iloc[0:0].copy() + + +def _precompute_feature_history(price_groups: Mapping[str, pd.DataFrame]) -> dict[str, pd.DataFrame]: + feature_history: dict[str, pd.DataFrame] = {} + for symbol, history in price_groups.items(): + closes = pd.to_numeric(history["close"], errors="coerce") + volumes = pd.to_numeric(history["volume"], errors="coerce") + returns = closes.pct_change() + ma20 = closes.rolling(20).mean() + ma50 = closes.rolling(50).mean() + ma200 = closes.rolling(200).mean() + rolling63max = closes.rolling(63).max() + rolling126max = closes.rolling(126).max() + rolling252max = closes.rolling(252).max() + drawdown_126 = closes / closes.rolling(126).max() - 1.0 + drawdown_126 = drawdown_126.replace([np.inf, -np.inf], np.nan) + maxdd_126 = drawdown_126.rolling(126).min() + feature_history[str(symbol).upper()] = pd.DataFrame( + { + "as_of": history["as_of"], + "close": closes, + "volume": volumes, + "adv20_usd": (closes * volumes).rolling(20).mean(), + "history_days": np.arange(1, len(history) + 1, dtype=int), + "mom_6_1": closes.shift(21) / closes.shift(147) - 1.0, + "mom_12_1": closes.shift(21) / closes.shift(273) - 1.0, + "sma20_gap": closes / ma20 - 1.0, + "sma50_gap": closes / ma50 - 1.0, + "sma200_gap": closes / ma200 - 1.0, + "ma50_over_ma200": ma50 / ma200 - 1.0, + "vol_63": returns.rolling(63).std(ddof=0) * math.sqrt(252), + "maxdd_126": maxdd_126, + "breakout_252": closes / rolling252max - 1.0, + "dist_63_high": closes / rolling63max - 1.0, + "dist_126_high": closes / rolling126max - 1.0, + "rebound_20": closes / closes.shift(20) - 1.0, + } + ) + return feature_history + + +def _lookup_features( + symbol: str, + as_of: pd.Timestamp, + feature_history_by_symbol: Mapping[str, pd.DataFrame], + *, + sector: str, +) -> dict[str, object]: + history = feature_history_by_symbol.get(str(symbol).upper()) + row = {"as_of": as_of, "symbol": symbol, "sector": sector} + if history is None or history.empty: + for column in FEATURE_SNAPSHOT_COLUMNS: + if column in row: + continue + row[column] = 0 if column == "history_days" else False if column == "base_eligible" else float("nan") + return row + + cutoff = int(history["as_of"].searchsorted(as_of, side="right")) + if cutoff <= 0: + for column in FEATURE_SNAPSHOT_COLUMNS: + if column in row: + continue + row[column] = 0 if column == "history_days" else False if column == "base_eligible" else float("nan") + return row + + current = history.iloc[cutoff - 1] + for column in FEATURE_SNAPSHOT_COLUMNS: + if column in row: + continue + if column == "base_eligible": + row[column] = False + continue + value = current[column] + if column == "history_days": + row[column] = int(value) if pd.notna(value) else 0 + else: + row[column] = float(value) if pd.notna(value) else float("nan") + return row + + +def build_feature_snapshot( + price_history, + universe_snapshot, + *, + as_of_date=None, + benchmark_symbol: str = BENCHMARK_SYMBOL, + benchmark_sector: str = "benchmark", + safe_haven: str = SAFE_HAVEN, + sector_whitelist: tuple[str, ...] = DEFAULT_SECTOR_WHITELIST, + min_price_usd: float = 10.0, + min_adv20_usd: float = DEFAULT_MIN_ADV20_USD, + min_history_days: int = 252, +) -> pd.DataFrame: + universe = pd.DataFrame(universe_snapshot).copy() + if universe.empty: + raise ValueError("universe_snapshot must contain at least one row") + _require_columns(universe, UNIVERSE_REQUIRED_COLUMNS, name="universe_snapshot") + + universe["symbol"] = _normalize_symbol_series(universe["symbol"]) + universe["sector"] = universe["sector"].fillna("unknown").astype(str).str.strip().replace("", "unknown") + if {"start_date", "end_date"} & set(universe.columns): + for column in ("start_date", "end_date"): + if column in universe.columns: + universe[column] = pd.to_datetime(universe[column], utc=False).map(_normalize_date) + if as_of_date is None: + raise ValueError(f"{PROFILE_NAME} requires as_of_date when universe history has start/end dates") + universe = resolve_active_universe(universe, as_of_date) + universe["symbol"] = _normalize_symbol_series(universe["symbol"]) + universe["sector"] = universe["sector"].fillna("unknown").astype(str).str.strip().replace("", "unknown") + + if sector_whitelist: + universe = universe.loc[universe["sector"].isin(tuple(sector_whitelist))].copy() + universe = universe.drop_duplicates(subset=["symbol"], keep="last") + + if as_of_date is None: + prices = pd.DataFrame(price_history) + if prices.empty or "as_of" not in prices.columns: + raise ValueError("price_history must contain at least one usable row") + as_of = pd.to_datetime(prices["as_of"], utc=False).map(_normalize_date).max() + else: + as_of = _normalize_date(as_of_date) + + price_groups, _ = _normalize_price_groups(price_history, as_of=as_of) + feature_history = _precompute_feature_history(price_groups) + + benchmark_symbol = str(benchmark_symbol or "").strip().upper() + safe_haven = str(safe_haven or "").strip().upper() + extra_symbols = [benchmark_symbol, "SPY", safe_haven] + sector_map = dict(zip(universe["symbol"], universe["sector"])) + symbols = universe["symbol"].tolist() + for extra in extra_symbols: + if extra and extra not in symbols: + symbols.append(extra) + + rows = [ + _lookup_features( + symbol, + as_of, + feature_history, + sector=sector_map.get(symbol, benchmark_sector if symbol in {benchmark_symbol, "SPY"} else "defense" if symbol == safe_haven else "unknown"), + ) + for symbol in symbols + ] + frame = pd.DataFrame(rows).sort_values("symbol").reset_index(drop=True) + + frame["base_eligible"] = ( + ~frame["symbol"].isin([benchmark_symbol, "SPY", safe_haven]) + & frame["history_days"].ge(min_history_days) + & frame["close"].gt(min_price_usd) + & frame["adv20_usd"].ge(min_adv20_usd) + & frame[ + [ + "mom_6_1", + "mom_12_1", + "sma20_gap", + "sma50_gap", + "sma200_gap", + "ma50_over_ma200", + "vol_63", + "maxdd_126", + "breakout_252", + "dist_63_high", + "dist_126_high", + "rebound_20", + ] + ].notna().all(axis=1) + ) + return frame.loc[:, FEATURE_SNAPSHOT_COLUMNS].reset_index(drop=True) diff --git a/src/us_equity_strategies/strategies/cash_buffer_branch_default.py b/src/us_equity_strategies/strategies/cash_buffer_branch_default.py new file mode 100644 index 00000000..a47ed623 --- /dev/null +++ b/src/us_equity_strategies/strategies/cash_buffer_branch_default.py @@ -0,0 +1,736 @@ +from __future__ import annotations + +import json +import math +from collections.abc import Mapping +from importlib import import_module +from pathlib import Path +from typing import Any + +import pandas as pd + +SIGNAL_SOURCE = "feature_snapshot" +STATUS_ICON = "🧲" +PROFILE_NAME = "cash_buffer_branch_default" +BRANCH_ROLE = "cash-buffered parallel branch" +BENCHMARK_SYMBOL = "QQQ" +SAFE_HAVEN = "BOXX" +DEFAULT_HOLDINGS_COUNT = 8 +DEFAULT_SINGLE_NAME_CAP = 0.10 +DEFAULT_SECTOR_CAP = 0.40 +DEFAULT_HOLD_BONUS = 0.10 +DEFAULT_RISK_ON_EXPOSURE = 0.80 +DEFAULT_SOFT_DEFENSE_EXPOSURE = 0.60 +DEFAULT_HARD_DEFENSE_EXPOSURE = 0.00 +DEFAULT_SOFT_BREADTH_THRESHOLD = 0.55 +DEFAULT_HARD_BREADTH_THRESHOLD = 0.35 +DEFAULT_MIN_ADV20_USD = 50_000_000.0 +DEFAULT_NORMALIZATION = "universe_cross_sectional" +DEFAULT_SCORE_TEMPLATE = "balanced_pullback" +DEFAULT_SECTOR_WHITELIST = ("Information Technology", "Communication") +DEFAULT_RUNTIME_EXECUTION_WINDOW_TRADING_DAYS = 3 +DEFAULT_EXECUTION_CASH_RESERVE_RATIO = 0.0 +SNAPSHOT_DATE_COLUMNS = ("as_of", "snapshot_date") +MAX_SNAPSHOT_MONTH_LAG = 1 +REQUIRE_SNAPSHOT_MANIFEST = True +SNAPSHOT_CONTRACT_VERSION = "cash_buffer_branch_default.feature_snapshot.v1" + +REQUIRED_FEATURE_COLUMNS = frozenset( + { + "symbol", + "sector", + "close", + "adv20_usd", + "history_days", + "mom_6_1", + "mom_12_1", + "sma20_gap", + "sma50_gap", + "sma200_gap", + "ma50_over_ma200", + "vol_63", + "maxdd_126", + "breakout_252", + "dist_63_high", + "dist_126_high", + "rebound_20", + } +) + +FEATURE_SIGNAL_KWARG_KEYS = ( + "benchmark_symbol", + "safe_haven", + "holdings_count", + "single_name_cap", + "sector_cap", + "hold_bonus", + "risk_on_exposure", + "soft_defense_exposure", + "hard_defense_exposure", + "soft_breadth_threshold", + "hard_breadth_threshold", + "min_adv20_usd", + "sector_whitelist", + "normalization", + "score_template", + "run_as_of", + "runtime_execution_window_trading_days", + "runtime_config_name", + "runtime_config_path", + "runtime_config_source", + "residual_proxy", +) + + +def _coerce_bool(value: Any) -> bool: + if pd.isna(value): + return False + if isinstance(value, bool): + return value + if isinstance(value, (int, float)): + return bool(value) + normalized = str(value).strip().lower() + if normalized in {"1", "true", "yes", "y"}: + return True + if normalized in {"0", "false", "no", "n"}: + return False + return bool(normalized) + + +def _normalize_symbol_series(values: pd.Series) -> pd.Series: + return values.astype(str).str.upper().str.strip() + + +def _normalize_holdings(current_holdings) -> set[str]: + if current_holdings is None: + return set() + raw_symbols = current_holdings.keys() if isinstance(current_holdings, Mapping) else current_holdings + normalized: set[str] = set() + for item in raw_symbols: + symbol = getattr(item, "symbol", item) + symbol_text = str(symbol or "").strip().upper() + if symbol_text: + normalized.add(symbol_text) + return normalized + + +def _to_frame(feature_snapshot) -> pd.DataFrame: + frame = feature_snapshot.copy() if isinstance(feature_snapshot, pd.DataFrame) else pd.DataFrame(list(feature_snapshot)) + if frame.empty: + raise ValueError("feature_snapshot must contain at least one row") + + missing = REQUIRED_FEATURE_COLUMNS - set(frame.columns) + if missing: + missing_text = ", ".join(sorted(missing)) + raise ValueError(f"feature_snapshot missing required columns: {missing_text}") + + frame["symbol"] = _normalize_symbol_series(frame["symbol"]) + frame["sector"] = frame["sector"].fillna("unknown").astype(str).str.strip().replace("", "unknown") + if "as_of" in frame.columns: + frame["as_of"] = pd.to_datetime(frame["as_of"], utc=False).dt.tz_localize(None).dt.normalize() + if "base_eligible" not in frame.columns: + if "eligible" in frame.columns: + frame["base_eligible"] = frame["eligible"] + else: + frame["base_eligible"] = True + frame["base_eligible"] = frame["base_eligible"].map(_coerce_bool) + + numeric_columns = REQUIRED_FEATURE_COLUMNS - {"symbol", "sector"} + for column in numeric_columns: + frame[column] = pd.to_numeric(frame[column], errors="coerce") + return frame + + +def _zscore(values: pd.Series) -> pd.Series: + numeric = pd.to_numeric(values, errors="coerce") + std = float(numeric.std(ddof=0)) + if pd.isna(std) or std == 0: + return pd.Series(0.0, index=values.index, dtype=float) + return ((numeric - numeric.mean()) / std).fillna(0.0) + + +def _group_zscore(values: pd.Series, group_keys: pd.Series | None) -> pd.Series: + if group_keys is None: + return _zscore(values) + numeric = pd.to_numeric(values, errors="coerce") + return numeric.groupby(group_keys).transform(_zscore).fillna(0.0) + + +def _apply_universe_filter( + frame: pd.DataFrame, + *, + benchmark_symbol: str, + safe_haven: str, + sector_whitelist: tuple[str, ...], + min_adv20_usd: float, +) -> pd.DataFrame: + filtered = frame.loc[ + ~frame["symbol"].isin([benchmark_symbol, safe_haven]) + & frame["base_eligible"] + & frame["adv20_usd"].ge(min_adv20_usd) + ].copy() + if sector_whitelist: + filtered = filtered.loc[filtered["sector"].isin(sector_whitelist)].copy() + return filtered + + +def _compute_family_features(scored: pd.DataFrame, benchmark_rows: pd.DataFrame) -> pd.DataFrame: + qqq_rows = benchmark_rows.loc[benchmark_rows["symbol"] == "QQQ"] + if qqq_rows.empty: + raise RuntimeError("QQQ benchmark row missing from snapshot") + qqq_row = qqq_rows.iloc[-1] + qqq_mom_6_1 = float(qqq_row["mom_6_1"]) + qqq_mom_12_1 = float(qqq_row["mom_12_1"]) + + scored = scored.copy() + scored["excess_mom_6_1"] = scored["mom_6_1"] - qqq_mom_6_1 + scored["excess_mom_12_1"] = scored["mom_12_1"] - qqq_mom_12_1 + scored["drawdown_abs"] = scored["maxdd_126"].abs() + scored["trend_strength"] = ( + scored["sma200_gap"] * 0.45 + + scored["breakout_252"] * 0.35 + + scored["ma50_over_ma200"] * 0.20 + ) + scored["controlled_pullback_score"] = ( + -((scored["dist_63_high"] + 0.08).abs() * 0.55) + -((scored["dist_126_high"] + 0.12).abs() * 0.25) + -(((-scored["sma50_gap"]).clip(lower=0.0)) * 0.10) + -(((-scored["sma200_gap"]).clip(lower=0.0)) * 0.10) + ) + scored["recovery_confirmation"] = ( + scored["sma20_gap"] * 0.40 + + scored["sma50_gap"] * 0.35 + + scored["rebound_20"] * 0.25 + ) + if scored["sector"].nunique() > 1: + group_median = scored.groupby("sector")["excess_mom_12_1"].transform("median") + else: + group_median = pd.Series(float(scored["excess_mom_12_1"].median()), index=scored.index) + scored["rel_strength_vs_group"] = scored["excess_mom_12_1"] - group_median + return scored + + +def _score_candidates( + frame: pd.DataFrame, + current_holdings: set[str], + *, + benchmark_symbol: str, + safe_haven: str, + sector_whitelist: tuple[str, ...], + min_adv20_usd: float, + normalization: str, + score_template: str, + hold_bonus: float, +) -> pd.DataFrame: + benchmark_rows = frame.loc[frame["symbol"].isin([benchmark_symbol, "SPY", "QQQ", "XLK", "SMH"])].copy() + eligible = _apply_universe_filter( + frame, + benchmark_symbol=benchmark_symbol, + safe_haven=safe_haven, + sector_whitelist=sector_whitelist, + min_adv20_usd=min_adv20_usd, + ) + if eligible.empty: + return eligible + scored = _compute_family_features(eligible, benchmark_rows) + + if normalization == "sector": + group_keys = scored["sector"] if scored["sector"].nunique() > 1 else None + elif normalization in {"universe", "universe_cross_sectional"}: + group_keys = None + else: + raise ValueError(f"Unsupported normalization: {normalization}") + + for column in ( + "excess_mom_12_1", + "excess_mom_6_1", + "trend_strength", + "controlled_pullback_score", + "recovery_confirmation", + "rel_strength_vs_group", + "vol_63", + "drawdown_abs", + ): + scored[f"z_{column}"] = _group_zscore(scored[column], group_keys) + + if score_template != "balanced_pullback": + raise ValueError(f"Unsupported score_template: {score_template}") + + scored["score"] = ( + scored["z_excess_mom_12_1"] * 0.25 + + scored["z_excess_mom_6_1"] * 0.20 + + scored["z_trend_strength"] * 0.15 + + scored["z_controlled_pullback_score"] * 0.15 + + scored["z_recovery_confirmation"] * 0.10 + + scored["z_rel_strength_vs_group"] * 0.10 + - scored["z_vol_63"] * 0.03 + - scored["z_drawdown_abs"] * 0.02 + ) + if current_holdings: + scored.loc[scored["symbol"].isin(current_holdings), "score"] += float(hold_bonus) + return scored + + +def _resolve_regime( + *, + benchmark_trend_positive: bool, + breadth_ratio: float, + soft_breadth_threshold: float, + hard_breadth_threshold: float, +) -> str: + if (not benchmark_trend_positive) and breadth_ratio < hard_breadth_threshold: + return "hard_defense" + if (not benchmark_trend_positive) or breadth_ratio < soft_breadth_threshold: + return "soft_defense" + return "risk_on" + + +def _stock_exposure_for_regime( + regime: str, + *, + risk_on_exposure: float, + soft_defense_exposure: float, + hard_defense_exposure: float, +) -> float: + if regime == "hard_defense": + return float(hard_defense_exposure) + if regime == "soft_defense": + return float(soft_defense_exposure) + return float(risk_on_exposure) + + +def build_target_weights( + feature_snapshot, + current_holdings, + *, + benchmark_symbol: str = BENCHMARK_SYMBOL, + safe_haven: str = SAFE_HAVEN, + holdings_count: int = DEFAULT_HOLDINGS_COUNT, + single_name_cap: float = DEFAULT_SINGLE_NAME_CAP, + sector_cap: float = DEFAULT_SECTOR_CAP, + hold_bonus: float = DEFAULT_HOLD_BONUS, + risk_on_exposure: float = DEFAULT_RISK_ON_EXPOSURE, + soft_defense_exposure: float = DEFAULT_SOFT_DEFENSE_EXPOSURE, + hard_defense_exposure: float = DEFAULT_HARD_DEFENSE_EXPOSURE, + soft_breadth_threshold: float = DEFAULT_SOFT_BREADTH_THRESHOLD, + hard_breadth_threshold: float = DEFAULT_HARD_BREADTH_THRESHOLD, + min_adv20_usd: float = DEFAULT_MIN_ADV20_USD, + sector_whitelist: tuple[str, ...] = DEFAULT_SECTOR_WHITELIST, + normalization: str = DEFAULT_NORMALIZATION, + score_template: str = DEFAULT_SCORE_TEMPLATE, + residual_proxy: str = "simple_excess_return_vs_QQQ", + runtime_config_name: str | None = None, + runtime_config_path: str | None = None, + runtime_config_source: str | None = None, +): + if holdings_count <= 0: + raise ValueError("holdings_count must be positive") + if single_name_cap <= 0 or sector_cap <= 0: + raise ValueError("single_name_cap and sector_cap must be positive") + + benchmark_symbol = str(benchmark_symbol or "").strip().upper() + safe_haven = str(safe_haven or "").strip().upper() + frame = _to_frame(feature_snapshot) + current_holdings_set = _normalize_holdings(current_holdings) + + benchmark_rows = frame.loc[frame["symbol"] == benchmark_symbol] + benchmark_trend_positive = True + if not benchmark_rows.empty: + benchmark_trend_positive = bool(float(benchmark_rows.iloc[-1]["sma200_gap"]) > 0) + + eligible_for_breadth = _apply_universe_filter( + frame, + benchmark_symbol=benchmark_symbol, + safe_haven=safe_haven, + sector_whitelist=tuple(sector_whitelist or ()), + min_adv20_usd=float(min_adv20_usd), + ) + breadth_ratio = float((eligible_for_breadth["sma200_gap"] > 0).mean()) if not eligible_for_breadth.empty else 0.0 + regime = _resolve_regime( + benchmark_trend_positive=benchmark_trend_positive, + breadth_ratio=breadth_ratio, + soft_breadth_threshold=float(soft_breadth_threshold), + hard_breadth_threshold=float(hard_breadth_threshold), + ) + stock_exposure = _stock_exposure_for_regime( + regime, + risk_on_exposure=float(risk_on_exposure), + soft_defense_exposure=float(soft_defense_exposure), + hard_defense_exposure=float(hard_defense_exposure), + ) + + if eligible_for_breadth.empty or stock_exposure <= 0: + signal = ( + f"regime={regime} breadth={breadth_ratio:.1%} " + f"benchmark_trend={'up' if benchmark_trend_positive else 'down'}" + ) + return {safe_haven: 1.0}, signal, { + "benchmark_symbol": benchmark_symbol, + "benchmark_trend_positive": benchmark_trend_positive, + "breadth_ratio": breadth_ratio, + "regime": regime, + "target_stock_weight": 0.0, + "realized_stock_weight": 0.0, + "safe_haven_weight": 1.0, + "selected_symbols": (), + "selected_count": 0, + "candidate_count": int(len(eligible_for_breadth)), + "runtime_config_name": runtime_config_name, + "runtime_config_path": runtime_config_path, + "runtime_config_source": runtime_config_source, + "residual_proxy": residual_proxy, + } + + scored = _score_candidates( + frame, + current_holdings_set, + benchmark_symbol=benchmark_symbol, + safe_haven=safe_haven, + sector_whitelist=tuple(sector_whitelist or ()), + min_adv20_usd=float(min_adv20_usd), + normalization=normalization, + score_template=score_template, + hold_bonus=float(hold_bonus), + ) + if scored.empty: + signal = ( + f"regime={regime} breadth={breadth_ratio:.1%} " + f"benchmark_trend={'up' if benchmark_trend_positive else 'down'} no_selection" + ) + return {safe_haven: 1.0}, signal, { + "benchmark_symbol": benchmark_symbol, + "benchmark_trend_positive": benchmark_trend_positive, + "breadth_ratio": breadth_ratio, + "regime": regime, + "target_stock_weight": 0.0, + "realized_stock_weight": 0.0, + "safe_haven_weight": 1.0, + "selected_symbols": (), + "selected_count": 0, + "candidate_count": 0, + "runtime_config_name": runtime_config_name, + "runtime_config_path": runtime_config_path, + "runtime_config_source": runtime_config_source, + "residual_proxy": residual_proxy, + } + + ranked = scored.sort_values( + by=["score", "excess_mom_12_1", "trend_strength", "symbol"], + ascending=[False, False, False, True], + ) + + per_name_target = stock_exposure / max(holdings_count, 1) + sector_slot_cap = holdings_count if per_name_target <= 0 else max(1, int(math.floor(float(sector_cap) / per_name_target))) + selected_rows = [] + sector_counts: dict[str, int] = {} + for row in ranked.itertuples(index=False): + sector = str(row.sector) + if sector_counts.get(sector, 0) >= sector_slot_cap: + continue + selected_rows.append(row._asdict()) + sector_counts[sector] = sector_counts.get(sector, 0) + 1 + if len(selected_rows) >= holdings_count: + break + selected = pd.DataFrame(selected_rows) + if selected.empty: + return {safe_haven: 1.0}, "no_selection", { + "benchmark_symbol": benchmark_symbol, + "benchmark_trend_positive": benchmark_trend_positive, + "breadth_ratio": breadth_ratio, + "regime": regime, + "target_stock_weight": 0.0, + "realized_stock_weight": 0.0, + "safe_haven_weight": 1.0, + "selected_symbols": (), + "selected_count": 0, + "candidate_count": int(len(scored)), + "sector_slot_cap": sector_slot_cap, + "runtime_config_name": runtime_config_name, + "runtime_config_path": runtime_config_path, + "runtime_config_source": runtime_config_source, + "residual_proxy": residual_proxy, + } + + per_name_weight = min(float(single_name_cap), stock_exposure / len(selected)) + invested_weight = float(per_name_weight * len(selected)) + safe_haven_weight = max(0.0, float(1.0 - invested_weight)) + weights = {row.symbol: float(per_name_weight) for row in selected.itertuples(index=False)} + if safe_haven_weight > 1e-12: + weights[safe_haven] = safe_haven_weight + + top_preview = ", ".join( + f"{row.symbol}({row.score:.2f})" + for row in selected.head(5).itertuples(index=False) + ) + signal = ( + f"regime={regime} breadth={breadth_ratio:.1%} " + f"benchmark_trend={'up' if benchmark_trend_positive else 'down'} " + f"target_stock={stock_exposure:.1%} realized_stock={invested_weight:.1%} " + f"selected={len(selected)} top={top_preview}" + ) + metadata = { + "benchmark_symbol": benchmark_symbol, + "benchmark_trend_positive": benchmark_trend_positive, + "breadth_ratio": breadth_ratio, + "regime": regime, + "target_stock_weight": float(stock_exposure), + "realized_stock_weight": invested_weight, + "safe_haven_weight": safe_haven_weight, + "selected_symbols": tuple(selected["symbol"].tolist()), + "selected_count": int(len(selected)), + "candidate_count": int(len(scored)), + "sector_slot_cap": sector_slot_cap, + "runtime_config_name": runtime_config_name, + "runtime_config_path": runtime_config_path, + "runtime_config_source": runtime_config_source, + "residual_proxy": residual_proxy, + } + return weights, signal, metadata + + +def extract_managed_symbols( + feature_snapshot, + *, + benchmark_symbol: str = BENCHMARK_SYMBOL, + safe_haven: str = SAFE_HAVEN, +) -> tuple[str, ...]: + frame = _to_frame(feature_snapshot) + benchmark_symbol = str(benchmark_symbol or "").strip().upper() + safe_haven = str(safe_haven or "").strip().upper() + symbols = [] + for symbol in frame["symbol"].tolist(): + if symbol == benchmark_symbol: + continue + symbols.append(symbol) + if safe_haven and safe_haven not in symbols: + symbols.append(safe_haven) + return tuple(dict.fromkeys(symbols)) + + +def _load_nyse_calendar() -> tuple[Any | None, str]: + try: + module = import_module("pandas_market_calendars") + except Exception: + return None, "business_day_fallback" + try: + calendar = module.get_calendar("NYSE") + except Exception: + return None, "business_day_fallback" + if calendar is None: + return None, "business_day_fallback" + return calendar, "nyse_calendar" + + +def _next_trading_days(after_date: pd.Timestamp, *, count: int) -> tuple[tuple[pd.Timestamp, ...], str]: + if count <= 0: + return (), "disabled" + start_date = pd.Timestamp(after_date).normalize() + pd.Timedelta(days=1) + calendar, calendar_source = _load_nyse_calendar() + if calendar is None: + return tuple(pd.bdate_range(start=start_date, periods=count).normalize()), calendar_source + end_date = start_date + pd.Timedelta(days=max(10, count * 5)) + schedule = calendar.schedule(start_date=start_date, end_date=end_date) + if getattr(schedule, "index", None) is None or len(schedule.index) == 0: + return (), calendar_source + sessions = pd.to_datetime(schedule.index) + if getattr(sessions, "tz", None) is not None: + sessions = sessions.tz_localize(None) + sessions = sessions.normalize() + return tuple(sessions[:count]), calendar_source + + +def evaluate_execution_window( + feature_snapshot, + *, + run_as_of=None, + runtime_execution_window_trading_days: int = DEFAULT_RUNTIME_EXECUTION_WINDOW_TRADING_DAYS, +) -> dict[str, object]: + frame = _to_frame(feature_snapshot) + if "as_of" not in frame.columns: + return { + "should_execute": True, + "no_op_reason": None, + "snapshot_as_of": None, + "execution_window": (), + } + + snapshot_as_of = pd.Timestamp(frame["as_of"].max()).normalize() + if run_as_of is None: + return { + "should_execute": True, + "no_op_reason": None, + "snapshot_as_of": snapshot_as_of, + "execution_window": (), + } + + run_date = pd.Timestamp(run_as_of).normalize() + allowed_days, calendar_source = _next_trading_days( + snapshot_as_of, + count=int(runtime_execution_window_trading_days), + ) + if not allowed_days: + return { + "should_execute": False, + "no_op_reason": f"no_execution_window_after_snapshot:{snapshot_as_of.date()}", + "snapshot_as_of": snapshot_as_of, + "execution_window": (), + "calendar_source": calendar_source, + } + if run_date not in allowed_days: + allowed_text = ",".join(day.date().isoformat() for day in allowed_days) + return { + "should_execute": False, + "no_op_reason": f"outside_monthly_execution_window snapshot={snapshot_as_of.date()} allowed={allowed_text}", + "snapshot_as_of": snapshot_as_of, + "execution_window": tuple(day.date().isoformat() for day in allowed_days), + "calendar_source": calendar_source, + } + return { + "should_execute": True, + "no_op_reason": None, + "snapshot_as_of": snapshot_as_of, + "execution_window": tuple(day.date().isoformat() for day in allowed_days), + "calendar_source": calendar_source, + } + + +def load_runtime_parameters( + *, + config_path: str | Path | None = None, + logger=None, +) -> dict[str, object]: + if logger is None: + logger = lambda message: None + + runtime_params = { + "benchmark_symbol": BENCHMARK_SYMBOL, + "safe_haven": SAFE_HAVEN, + "holdings_count": DEFAULT_HOLDINGS_COUNT, + "single_name_cap": DEFAULT_SINGLE_NAME_CAP, + "sector_cap": DEFAULT_SECTOR_CAP, + "hold_bonus": DEFAULT_HOLD_BONUS, + "risk_on_exposure": DEFAULT_RISK_ON_EXPOSURE, + "soft_defense_exposure": DEFAULT_SOFT_DEFENSE_EXPOSURE, + "hard_defense_exposure": DEFAULT_HARD_DEFENSE_EXPOSURE, + "soft_breadth_threshold": DEFAULT_SOFT_BREADTH_THRESHOLD, + "hard_breadth_threshold": DEFAULT_HARD_BREADTH_THRESHOLD, + "min_adv20_usd": DEFAULT_MIN_ADV20_USD, + "sector_whitelist": DEFAULT_SECTOR_WHITELIST, + "normalization": DEFAULT_NORMALIZATION, + "score_template": DEFAULT_SCORE_TEMPLATE, + "runtime_execution_window_trading_days": DEFAULT_RUNTIME_EXECUTION_WINDOW_TRADING_DAYS, + "execution_cash_reserve_ratio": DEFAULT_EXECUTION_CASH_RESERVE_RATIO, + "runtime_config_name": PROFILE_NAME, + "runtime_config_path": None, + "runtime_config_source": "module_defaults", + "residual_proxy": "simple_excess_return_vs_QQQ", + } + if config_path is None: + logger(f"[{PROFILE_NAME}] runtime config source=module_defaults") + return runtime_params + + config_file = Path(config_path) + if not config_file.exists(): + raise FileNotFoundError(f"Runtime strategy config not found: {config_file}") + payload = json.loads(config_file.read_text(encoding="utf-8")) + if str(payload.get("name")).strip() != PROFILE_NAME: + raise ValueError(f"Runtime config name must be {PROFILE_NAME!r}") + if str(payload.get("family")).strip() != "tech_heavy_pullback": + raise ValueError("Runtime config family must be 'tech_heavy_pullback'") + if str(payload.get("branch_role")).strip() != BRANCH_ROLE: + raise ValueError(f"Runtime config branch_role must be {BRANCH_ROLE!r}") + + exposures = payload.get("exposures") or {} + breadth_thresholds = payload.get("breadth_thresholds") or {} + runtime_params.update( + { + "benchmark_symbol": str(payload.get("benchmark_symbol") or BENCHMARK_SYMBOL).upper(), + "holdings_count": int(payload.get("holdings_count", DEFAULT_HOLDINGS_COUNT)), + "single_name_cap": float(payload.get("single_name_cap", DEFAULT_SINGLE_NAME_CAP)), + "sector_cap": float(payload.get("sector_cap", DEFAULT_SECTOR_CAP)), + "hold_bonus": float(payload.get("hold_bonus", DEFAULT_HOLD_BONUS)), + "risk_on_exposure": float(exposures.get("risk_on", DEFAULT_RISK_ON_EXPOSURE)), + "soft_defense_exposure": float(exposures.get("soft_defense", DEFAULT_SOFT_DEFENSE_EXPOSURE)), + "hard_defense_exposure": float(exposures.get("hard_defense", DEFAULT_HARD_DEFENSE_EXPOSURE)), + "soft_breadth_threshold": float(breadth_thresholds.get("soft", DEFAULT_SOFT_BREADTH_THRESHOLD)), + "hard_breadth_threshold": float(breadth_thresholds.get("hard", DEFAULT_HARD_BREADTH_THRESHOLD)), + "min_adv20_usd": float(payload.get("min_adv20_usd", DEFAULT_MIN_ADV20_USD)), + "sector_whitelist": tuple(payload.get("sector_whitelist") or DEFAULT_SECTOR_WHITELIST), + "normalization": str(payload.get("normalization") or DEFAULT_NORMALIZATION), + "score_template": str(payload.get("score_template") or DEFAULT_SCORE_TEMPLATE), + "execution_cash_reserve_ratio": float( + payload.get("execution_cash_reserve_ratio", DEFAULT_EXECUTION_CASH_RESERVE_RATIO) + ), + "runtime_config_name": str(payload.get("name") or PROFILE_NAME), + "runtime_config_path": str(config_file), + "runtime_config_source": "external_config", + "residual_proxy": str(payload.get("residual_proxy") or "simple_excess_return_vs_QQQ"), + } + ) + logger(f"[{PROFILE_NAME}] runtime config source=external_config path={config_file}") + return runtime_params + + +def compute_signals( + feature_snapshot, + current_holdings, + *, + run_as_of=None, + runtime_execution_window_trading_days: int = DEFAULT_RUNTIME_EXECUTION_WINDOW_TRADING_DAYS, + **kwargs, +): + managed_symbols = extract_managed_symbols( + feature_snapshot, + benchmark_symbol=kwargs.get("benchmark_symbol", BENCHMARK_SYMBOL), + safe_haven=kwargs.get("safe_haven", SAFE_HAVEN), + ) + execution_window = evaluate_execution_window( + feature_snapshot, + run_as_of=run_as_of, + runtime_execution_window_trading_days=runtime_execution_window_trading_days, + ) + if not execution_window["should_execute"]: + status_desc = ( + f"no-op | reason={execution_window['no_op_reason']} | " + f"snapshot_as_of={execution_window['snapshot_as_of']}" + ) + return ( + None, + "monthly snapshot cadence | waiting inside execution window", + False, + status_desc, + { + "managed_symbols": managed_symbols, + "status_icon": STATUS_ICON, + "snapshot_as_of": execution_window["snapshot_as_of"], + "execution_window": execution_window["execution_window"], + "no_op_reason": execution_window["no_op_reason"], + "execution_calendar_source": execution_window.get("calendar_source"), + }, + ) + + weights, signal_desc, metadata = build_target_weights( + feature_snapshot, + current_holdings, + **kwargs, + ) + status_desc = ( + f"regime={metadata['regime']} | " + f"breadth={metadata['breadth_ratio']:.1%} | " + f"target_stock={metadata['target_stock_weight']:.1%} | " + f"realized_stock={metadata['realized_stock_weight']:.1%}" + ) + return ( + weights, + signal_desc, + metadata["regime"] == "hard_defense", + status_desc, + { + **metadata, + "managed_symbols": managed_symbols, + "status_icon": STATUS_ICON, + "snapshot_as_of": execution_window["snapshot_as_of"], + "execution_window": execution_window["execution_window"], + "execution_calendar_source": execution_window.get("calendar_source"), + }, + ) diff --git a/tests/test_cash_buffer_branch_default.py b/tests/test_cash_buffer_branch_default.py new file mode 100644 index 00000000..efa59aca --- /dev/null +++ b/tests/test_cash_buffer_branch_default.py @@ -0,0 +1,166 @@ +from __future__ import annotations + +import json +import unittest +from pathlib import Path +from tempfile import TemporaryDirectory + +import pandas as pd + + +def _feature_snapshot() -> pd.DataFrame: + as_of = pd.Timestamp("2026-03-31") + rows = [ + { + "as_of": as_of, + "symbol": "QQQ", + "sector": "benchmark", + "close": 500.0, + "volume": 1_000_000, + "adv20_usd": 1_000_000_000.0, + "history_days": 400, + "mom_6_1": 0.20, + "mom_12_1": 0.30, + "sma20_gap": 0.03, + "sma50_gap": 0.05, + "sma200_gap": 0.08, + "ma50_over_ma200": 0.04, + "vol_63": 0.22, + "maxdd_126": -0.12, + "breakout_252": -0.01, + "dist_63_high": -0.03, + "dist_126_high": -0.05, + "rebound_20": 0.04, + "base_eligible": False, + }, + { + "as_of": as_of, + "symbol": "BOXX", + "sector": "defense", + "close": 101.0, + "volume": 200_000, + "adv20_usd": 20_000_000.0, + "history_days": 400, + "mom_6_1": 0.02, + "mom_12_1": 0.04, + "sma20_gap": 0.00, + "sma50_gap": 0.00, + "sma200_gap": 0.01, + "ma50_over_ma200": 0.00, + "vol_63": 0.03, + "maxdd_126": -0.01, + "breakout_252": 0.00, + "dist_63_high": -0.01, + "dist_126_high": -0.01, + "rebound_20": 0.00, + "base_eligible": False, + }, + ] + tech_rows = [ + ("AAPL", "Information Technology", 0.18, 0.31, 0.02, 0.04, 0.07, 0.03, 0.19, -0.10, -0.02, -0.04, -0.07, 0.05), + ("MSFT", "Information Technology", 0.16, 0.29, 0.02, 0.04, 0.06, 0.03, 0.18, -0.11, -0.03, -0.05, -0.08, 0.04), + ("NVDA", "Information Technology", 0.28, 0.55, 0.05, 0.08, 0.16, 0.09, 0.32, -0.01, -0.02, -0.04, -0.06, 0.10), + ("META", "Communication", 0.20, 0.38, 0.03, 0.05, 0.10, 0.05, 0.25, -0.06, -0.04, -0.08, -0.10, 0.06), + ("GOOGL", "Communication", 0.17, 0.27, 0.02, 0.03, 0.07, 0.03, 0.20, -0.08, -0.05, -0.09, -0.11, 0.05), + ("NFLX", "Communication", 0.19, 0.34, 0.03, 0.05, 0.09, 0.04, 0.22, -0.05, -0.03, -0.07, -0.09, 0.05), + ("TTWO", "Communication", 0.14, 0.21, 0.01, 0.02, 0.05, 0.02, 0.14, -0.06, -0.04, -0.08, -0.11, 0.03), + ("CRM", "Information Technology", 0.14, 0.24, 0.01, 0.02, 0.05, 0.02, 0.16, -0.07, -0.05, -0.08, -0.10, 0.03), + ("ADBE", "Information Technology", 0.13, 0.22, 0.01, 0.02, 0.04, 0.01, 0.15, -0.09, -0.05, -0.07, -0.09, 0.02), + ("NOW", "Information Technology", 0.15, 0.26, 0.02, 0.03, 0.05, 0.02, 0.18, -0.04, -0.03, -0.06, -0.09, 0.05), + ] + for symbol, sector, mom6, mom12, sma20, sma50, sma200, ma50_200, breakout, d63, d126, mdd, vol, rebound in tech_rows: + rows.append( + { + "as_of": as_of, + "symbol": symbol, + "sector": sector, + "close": 100.0, + "volume": 1_000_000, + "adv20_usd": 120_000_000.0, + "history_days": 400, + "mom_6_1": mom6, + "mom_12_1": mom12, + "sma20_gap": sma20, + "sma50_gap": sma50, + "sma200_gap": sma200, + "ma50_over_ma200": ma50_200, + "vol_63": vol, + "maxdd_126": d126, + "breakout_252": breakout, + "dist_63_high": d63, + "dist_126_high": d126, + "rebound_20": rebound, + "base_eligible": True, + } + ) + return pd.DataFrame(rows) + + +class CashBufferBranchDefaultStrategyTest(unittest.TestCase): + def test_build_target_weights_is_geometry_honest(self): + from us_equity_strategies.strategies.cash_buffer_branch_default import build_target_weights + + weights, signal, metadata = build_target_weights( + _feature_snapshot(), + current_holdings={"AAPL"}, + ) + + self.assertIn("target_stock=80.0%", signal) + self.assertAlmostEqual(sum(weights.values()), 1.0, places=8) + self.assertAlmostEqual(metadata["target_stock_weight"], 0.8, places=8) + self.assertAlmostEqual(metadata["realized_stock_weight"], 0.8, places=8) + self.assertEqual(len(metadata["selected_symbols"]), 8) + self.assertAlmostEqual(weights["BOXX"], 0.2, places=8) + + def test_compute_signals_noops_outside_execution_window(self): + from us_equity_strategies.strategies.cash_buffer_branch_default import compute_signals + + weights, _signal, _emergency, status_desc, metadata = compute_signals( + _feature_snapshot(), + current_holdings=set(), + run_as_of="2026-04-10", + ) + + self.assertIsNone(weights) + self.assertIn("outside_monthly_execution_window", metadata["no_op_reason"]) + self.assertIn("no-op", status_desc) + + def test_load_runtime_parameters_reads_canonical_config(self): + from us_equity_strategies.strategies.cash_buffer_branch_default import load_runtime_parameters + + with TemporaryDirectory() as tmp_dir: + config_path = Path(tmp_dir) / "cash_buffer_branch_default.json" + config_path.write_text( + json.dumps( + { + "name": "cash_buffer_branch_default", + "family": "tech_heavy_pullback", + "branch_role": "cash-buffered parallel branch", + "benchmark_symbol": "QQQ", + "holdings_count": 8, + "single_name_cap": 0.10, + "sector_cap": 0.40, + "hold_bonus": 0.10, + "min_adv20_usd": 50_000_000.0, + "normalization": "universe_cross_sectional", + "score_template": "balanced_pullback", + "sector_whitelist": ["Information Technology", "Communication"], + "residual_proxy": "simple_excess_return_vs_QQQ", + "breadth_thresholds": {"soft": 0.55, "hard": 0.35}, + "exposures": {"risk_on": 0.8, "soft_defense": 0.6, "hard_defense": 0.0}, + "execution_cash_reserve_ratio": 0.0, + } + ), + encoding="utf-8", + ) + + params = load_runtime_parameters(config_path=config_path) + + self.assertEqual(params["runtime_config_source"], "external_config") + self.assertEqual(params["runtime_config_name"], "cash_buffer_branch_default") + self.assertEqual(params["sector_whitelist"], ("Information Technology", "Communication")) + self.assertEqual(params["execution_cash_reserve_ratio"], 0.0) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_cash_buffer_branch_feature_snapshot.py b/tests/test_cash_buffer_branch_feature_snapshot.py new file mode 100644 index 00000000..028f9c61 --- /dev/null +++ b/tests/test_cash_buffer_branch_feature_snapshot.py @@ -0,0 +1,97 @@ +from __future__ import annotations + +import unittest +from pathlib import Path +from tempfile import TemporaryDirectory +import json + +import pandas as pd + + +def _price_frame() -> pd.DataFrame: + start = pd.Timestamp("2025-01-01") + rows = [] + for day in range(420): + as_of = start + pd.Timedelta(days=day) + rows.append({"symbol": "AAPL", "as_of": as_of, "close": 150.0 + day * 0.20, "volume": 2_000_000}) + rows.append({"symbol": "MSFT", "as_of": as_of, "close": 300.0 + day * 0.15, "volume": 1_500_000}) + rows.append({"symbol": "META", "as_of": as_of, "close": 200.0 + day * 0.18, "volume": 1_200_000}) + rows.append({"symbol": "JNJ", "as_of": as_of, "close": 160.0 + day * 0.02, "volume": 800_000}) + rows.append({"symbol": "QQQ", "as_of": as_of, "close": 400.0 + day * 0.25, "volume": 3_000_000}) + rows.append({"symbol": "SPY", "as_of": as_of, "close": 500.0 + day * 0.20, "volume": 4_000_000}) + rows.append({"symbol": "BOXX", "as_of": as_of, "close": 101.0 + day * 0.005, "volume": 250_000}) + return pd.DataFrame(rows) + + +class CashBufferBranchFeatureSnapshotTest(unittest.TestCase): + def test_build_feature_snapshot_filters_to_tech_sectors(self): + from us_equity_strategies.snapshots.cash_buffer_branch_default import build_feature_snapshot + + snapshot = build_feature_snapshot( + _price_frame(), + [ + {"symbol": "AAPL", "sector": "Information Technology"}, + {"symbol": "MSFT", "sector": "Information Technology"}, + {"symbol": "META", "sector": "Communication"}, + {"symbol": "JNJ", "sector": "Health Care"}, + ], + as_of_date="2026-02-24", + ) + + symbols = set(snapshot["symbol"]) + self.assertIn("AAPL", symbols) + self.assertIn("META", symbols) + self.assertNotIn("JNJ", symbols) + self.assertIn("QQQ", symbols) + self.assertIn("BOXX", symbols) + base_flags = dict(zip(snapshot["symbol"], snapshot["base_eligible"])) + self.assertTrue(base_flags["AAPL"]) + self.assertFalse(base_flags["QQQ"]) + self.assertFalse(base_flags["BOXX"]) + + def test_cli_writes_snapshot(self): + from scripts.generate_cash_buffer_branch_feature_snapshot import main + + with TemporaryDirectory() as tmp_dir: + tmp_path = Path(tmp_dir) + prices_path = tmp_path / "prices.csv" + universe_path = tmp_path / "universe.csv" + output_path = tmp_path / "snapshot.csv" + config_path = tmp_path / "cash_buffer_branch_default.json" + + _price_frame().to_csv(prices_path, index=False) + universe_path.write_text( + "symbol,sector\nAAPL,Information Technology\nMSFT,Information Technology\nMETA,Communication\n", + encoding="utf-8", + ) + config_path.write_text( + json.dumps( + { + "name": "cash_buffer_branch_default", + } + ), + encoding="utf-8", + ) + + exit_code = main( + [ + "--prices", + str(prices_path), + "--universe", + str(universe_path), + "--output", + str(output_path), + "--config-path", + str(config_path), + "--as-of", + "2026-02-24", + ] + ) + + self.assertEqual(exit_code, 0) + self.assertTrue(output_path.exists()) + self.assertTrue(Path(f"{output_path}.manifest.json").exists()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_catalog.py b/tests/test_catalog.py index 930f29db..561d33e2 100644 --- a/tests/test_catalog.py +++ b/tests/test_catalog.py @@ -3,6 +3,7 @@ from quant_platform_kit.common.strategies import get_strategy_component_map from us_equity_strategies import get_strategy_definitions from us_equity_strategies.catalog import ( + CASH_BUFFER_BRANCH_DEFAULT_PROFILE, GLOBAL_ETF_ROTATION_PROFILE, HYBRID_GROWTH_INCOME_PROFILE, SEMICONDUCTOR_ROTATION_INCOME_PROFILE, @@ -25,6 +26,10 @@ def test_catalog_contains_supported_profiles(self): self.assertEqual(catalog[SEMICONDUCTOR_ROTATION_INCOME_PROFILE].domain, "us_equity") self.assertEqual(catalog[SEMICONDUCTOR_ROTATION_INCOME_PROFILE].supported_platforms, frozenset({"longbridge"})) + self.assertIn(CASH_BUFFER_BRANCH_DEFAULT_PROFILE, catalog) + self.assertEqual(catalog[CASH_BUFFER_BRANCH_DEFAULT_PROFILE].domain, "us_equity") + self.assertEqual(catalog[CASH_BUFFER_BRANCH_DEFAULT_PROFILE].supported_platforms, frozenset({"ibkr"})) + def test_known_profile_resolves(self): definition = get_strategy_definition("global_etf_rotation") self.assertEqual(definition.profile, GLOBAL_ETF_ROTATION_PROFILE) @@ -50,6 +55,14 @@ def test_known_profile_resolves(self): "us_equity_strategies.strategies.semiconductor_rotation_income", ) + cash_buffer_definition = get_strategy_definition("cash_buffer_branch_default") + self.assertEqual(cash_buffer_definition.profile, CASH_BUFFER_BRANCH_DEFAULT_PROFILE) + cash_buffer_module = get_strategy_component_map(cash_buffer_definition)["signal_logic"] + self.assertEqual( + cash_buffer_module.module_path, + "us_equity_strategies.strategies.cash_buffer_branch_default", + ) + if __name__ == "__main__": unittest.main() From ced79a225008df4e45bda9e6f076c30e1521e11e Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 5 Apr 2026 04:55:56 +0800 Subject: [PATCH 2/3] restore existing strategy support --- README.md | 298 ++++++++++++--- src/us_equity_strategies/catalog.py | 12 + .../snapshots/__init__.py | 14 +- .../russell_1000_multi_factor_defensive.py | 323 ++++++++++++++++ .../russell_1000_multi_factor_defensive.py | 361 ++++++++++++++++++ 5 files changed, 953 insertions(+), 55 deletions(-) create mode 100644 src/us_equity_strategies/snapshots/russell_1000_multi_factor_defensive.py create mode 100644 src/us_equity_strategies/strategies/russell_1000_multi_factor_defensive.py diff --git a/README.md b/README.md index d739990c..0d696bda 100644 --- a/README.md +++ b/README.md @@ -16,7 +16,7 @@ This repository is the strategy layer: it owns pure signal, allocation, and targ | Profile | Downstream runtime today | Core idea | | --- | --- | --- | | `global_etf_rotation` | `InteractiveBrokersPlatform` | Quarterly top-2 global ETF rotation with a daily canary defense | -| `cash_buffer_branch_default` | `InteractiveBrokersPlatform` | Tech-heavy monthly stock selection with an explicit 20% BOXX/cash buffer in risk-on and a QQQ+breadth defense ladder | +| `russell_1000_multi_factor_defensive` | `InteractiveBrokersPlatform` | Russell 1000 price-only monthly stock selection with SPY + breadth defense and BOXX parking | | `hybrid_growth_income` | `CharlesSchwabPlatform` | QQQ-driven TQQQ attack layer plus SPYI / QQQI income layer and BOXX defense | | `semiconductor_rotation_income` | `LongBridgePlatform` | SOXL / SOXX trend switch with BOXX parking and an additive income sleeve | @@ -51,25 +51,131 @@ These strategies are consumed by platform repositories through `QuantPlatformKit - Compared with a pure tech or leveraged-Nasdaq approach, this profile is meant to be steadier. - It still allows `VOO`, `XLK`, and `SMH` to win their way into the rotation instead of hard-coding them out. -### cash_buffer_branch_default +### russell_1000_multi_factor_defensive **Objective** -- Provide a research-only but runtime-loadable stock branch for concentrated tech leaders bought on controlled pullbacks. -- Keep the branch geometry honest: in `risk_on`, the profile explicitly targets `80%` stock exposure and parks the rest in `BOXX` / cash instead of relying on accidental underinvestment. +- Provide a first stock-level US equity strategy that stays close to the current platform architecture. +- Start with a price-only factor stack before adding fundamentals or ML reranking. +- Keep execution realistic by consuming a precomputed feature snapshot instead of fetching 1000 symbols live during the rebalance run. -**Current default shape** -- Universe: large-cap US tech / communication names from the shared snapshot task -- Benchmark: `QQQ` +**Universe** +- Point-in-time Russell 1000 constituent snapshot supplied by an upstream data task. +- Benchmark row: `SPY` - Safe haven: `BOXX` -- Position count: `8` -- Single-name cap: `10%` -- Sector cap: `40%` -- Default exposures: `80% / 60% / 0%` -**Current runtime contract** -- Consumes a precomputed feature snapshot plus a sidecar manifest -- Expects the canonical config name `cash_buffer_branch_default` -- Designed for monthly execution windows; non-window runs should no-op +**Signals and rules** +- Current V1 factors are price-only: + - `mom_6_1` + - `mom_12_1` + - `sma200_gap` + - `vol_63` + - `maxdd_126` +- Factors are standardized within sector, then combined into one total score. +- Existing holdings receive a configurable hold bonus. +- Market defense uses: + - `SPY` trend (`sma200_gap > 0`) + - breadth = share of eligible universe above `200MA` + +**Portfolio behavior** +- Rebalance cadence is monthly in the downstream runtime. +- Default stock exposure: + - `100%` in `risk_on` + - `50%` in `soft_defense` + - `10%` in `hard_defense` +- Default position count is `24`. +- Unused capital is parked in `BOXX`. + +**Feature snapshot schema** +- Required price-history input columns: + - `symbol`, `as_of`, `close`, `volume` +- Required universe input columns: + - `symbol`, `sector` + - optional: `start_date`, `end_date` for point-in-time membership during backtests +- Generated snapshot columns: + - `as_of`, `symbol`, `sector`, `close`, `volume`, `adv20_usd`, `history_days` + - `mom_6_1`, `mom_12_1`, `sma200_gap`, `vol_63`, `maxdd_126`, `eligible` + +**CLI task entry** + +Generate one snapshot directly: + +```bash +PYTHONPATH=src:. python3 scripts/generate_russell_1000_feature_snapshot.py \ + --prices /path/to/russell_1000_prices.csv \ + --universe /path/to/russell_1000_universe.csv \ + --output /path/to/r1000_feature_snapshot.csv \ + --benchmark-symbol SPY +``` + +Or run the env-driven wrapper task: + +```bash +export R1000_PRICE_HISTORY_PATH=/path/to/russell_1000_prices.csv +export R1000_UNIVERSE_PATH=/path/to/russell_1000_universe.csv +export R1000_FEATURE_SNAPSHOT_PATH=/path/to/r1000_feature_snapshot.csv +PYTHONPATH=src:. python3 scripts/run_russell_1000_snapshot_task.py +``` + +Starter sample inputs live in: + +- `examples/russell_1000_snapshot/universe.sample.csv` +- `examples/russell_1000_snapshot/prices.sample.csv` +- `examples/russell_1000_universe_snapshots/` + +**Minimal backtest entry** + +```bash +PYTHONPATH=src:. python3 scripts/backtest_russell_1000_multi_factor_defensive.py \ + --prices /path/to/russell_1000_prices.csv \ + --universe /path/to/russell_1000_universe.csv \ + --start 2018-01-01 \ + --end 2025-12-31 \ + --output-dir /path/to/backtest_outputs +``` + +The output directory will include: + +- `summary.csv` +- `portfolio_returns.csv` +- `weights_history.csv` +- `turnover_history.csv` + +**End-to-end local research workflow** + +1. Build interval-form universe history from dated constituent snapshots: + +```bash +PYTHONPATH=src:. python3 scripts/build_russell_1000_universe_history.py \ + --input-dir examples/russell_1000_universe_snapshots \ + --output /tmp/r1000_universe_history.csv +``` + +2. Fetch price history with Yahoo Finance: + +```bash +PYTHONPATH=src:. python3 scripts/fetch_russell_1000_price_history.py \ + --universe-history /tmp/r1000_universe_history.csv \ + --output /tmp/r1000_price_history.csv \ + --start 2024-01-01 +``` + +3. Generate a latest feature snapshot: + +```bash +PYTHONPATH=src:. python3 scripts/generate_russell_1000_feature_snapshot.py \ + --prices /tmp/r1000_price_history.csv \ + --universe /tmp/r1000_universe_history.csv \ + --output /tmp/r1000_feature_snapshot.csv +``` + +4. Run the backtest: + +```bash +PYTHONPATH=src:. python3 scripts/backtest_russell_1000_multi_factor_defensive.py \ + --prices /tmp/r1000_price_history.csv \ + --universe /tmp/r1000_universe_history.csv \ + --output-dir /tmp/r1000_backtest +``` ### hybrid_growth_income @@ -171,7 +277,7 @@ These strategies are consumed by platform repositories through `QuantPlatformKit | 策略档位 | 当前下游运行仓库 | 核心思路 | | --- | --- | --- | | `global_etf_rotation` | `InteractiveBrokersPlatform` | 22 只全球 ETF 的季度 Top 2 轮动，带每日 canary 防守 | -| `cash_buffer_branch_default` | `InteractiveBrokersPlatform` | 偏科技个股的月频受控回调分支，`risk_on` 明确只上 `80%` 股票，其余停在 `BOXX` / 现金 | +| `russell_1000_multi_factor_defensive` | `InteractiveBrokersPlatform` | Russell 1000 个股月频 price-only 选股，带 SPY + breadth 防守和 BOXX 停泊 | | `hybrid_growth_income` | `CharlesSchwabPlatform` | 由 QQQ 驱动的 TQQQ 攻击层，加上 SPYI / QQQI 收入层和 BOXX 防守层 | | `semiconductor_rotation_income` | `LongBridgePlatform` | SOXL / SOXX 趋势切换，剩余资金停在 BOXX，并叠加收入层 | @@ -206,45 +312,131 @@ These strategies are consumed by platform repositories through `QuantPlatformKit - 相比纯科技或者杠杆纳指路线，这个档位更稳。 - 但它仍然允许 `VOO`、`XLK`、`SMH` 靠表现进入组合，而不是事先把它们排除。 -### cash_buffer_branch_default - -**Objective** -- Provide a research-only but runtime-loadable stock branch for concentrated tech leaders bought on controlled pullbacks. -- Keep the branch geometry honest: in `risk_on`, the profile explicitly targets `80%` stock exposure and parks the rest in `BOXX` / cash instead of relying on accidental underinvestment. - -**Current default shape** -- Universe: large-cap US tech / communication names from the shared snapshot task -- Benchmark: `QQQ` -- Safe haven: `BOXX` -- Position count: `8` -- Single-name cap: `10%` -- Sector cap: `40%` -- Default exposures: `80% / 60% / 0%` - -**Current runtime contract** -- Consumes a precomputed feature snapshot plus a sidecar manifest -- Expects the canonical config name `cash_buffer_branch_default` -- Designed for monthly execution windows; non-window runs should no-op - -### cash_buffer_branch_default +### russell_1000_multi_factor_defensive **策略目标** -- 提供一条研究优先、但已经能被下游 runtime 正式加载的个股分支。 -- 核心是做偏科技龙头的受控回调买入，同时把 `risk_on` 的 `80%` 股票暴露显式写进规格里，不再靠隐式留仓位。 - -**当前默认规格** -- 股票池：共享 snapshot 任务里可交易的大盘科技 / 通信股票 -- 基准：`QQQ` -- 防守腿：`BOXX` -- 持仓数：`8` -- 单票上限：`10%` -- 行业上限：`40%` -- 默认暴露：`80% / 60% / 0%` - -**运行约定** -- 运行时消费预先生成好的 feature snapshot 和 sidecar manifest -- canonical config 名必须是 `cash_buffer_branch_default` -- 设计上保持月频执行，非执行窗口应显式 no-op +- 作为第一版个股策略，先尽量复用现有平台边界。 +- 第一阶段只用价格因子，不急着上基本面和机器学习。 +- 运行时只消费预先算好的 feature snapshot，不在调仓时现场拉 1000 只股票历史数据。 + +**股票池** +- 上游数据任务提供的 Russell 1000 点时成分快照 +- 基准行：`SPY` +- 防守资产：`BOXX` + +**当前 V1 因子** +- `mom_6_1` +- `mom_12_1` +- `sma200_gap` +- `vol_63` +- `maxdd_126` + +策略先在行业内做标准化，再合成总分。当前持仓可以拿到一小段 hold bonus。 + +**防守规则** +- `SPY` 的 `sma200_gap > 0` 代表 benchmark 趋势正常 +- breadth = 合格股票里站上 `200MA` 的比例 +- 默认风险暴露： + - `risk_on`：`100%` + - `soft_defense`：`50%` + - `hard_defense`：`10%` + +**组合规则** +- 下游运行时按月调仓 +- 默认持仓数 `24` +- 剩余资金停在 `BOXX` + +**feature snapshot 输入/输出约定** +- 价格历史输入列： + - `symbol`、`as_of`、`close`、`volume` +- 股票池输入列： + - `symbol`、`sector` + - 可选：`start_date`、`end_date`（用于回测时按日期启用 / 退出成分股） +- 生成后的 snapshot 列： + - `as_of`、`symbol`、`sector`、`close`、`volume`、`adv20_usd`、`history_days` + - `mom_6_1`、`mom_12_1`、`sma200_gap`、`vol_63`、`maxdd_126`、`eligible` + +**命令行任务入口** + +直接生成 snapshot： + +```bash +PYTHONPATH=src:. python3 scripts/generate_russell_1000_feature_snapshot.py \ + --prices /path/to/russell_1000_prices.csv \ + --universe /path/to/russell_1000_universe.csv \ + --output /path/to/r1000_feature_snapshot.csv \ + --benchmark-symbol SPY +``` + +或者用环境变量包装脚本： + +```bash +export R1000_PRICE_HISTORY_PATH=/path/to/russell_1000_prices.csv +export R1000_UNIVERSE_PATH=/path/to/russell_1000_universe.csv +export R1000_FEATURE_SNAPSHOT_PATH=/path/to/r1000_feature_snapshot.csv +PYTHONPATH=src:. python3 scripts/run_russell_1000_snapshot_task.py +``` + +示例输入文件： + +- `examples/russell_1000_snapshot/universe.sample.csv` +- `examples/russell_1000_snapshot/prices.sample.csv` +- `examples/russell_1000_universe_snapshots/` + +**最小回测入口** + +```bash +PYTHONPATH=src:. python3 scripts/backtest_russell_1000_multi_factor_defensive.py \ + --prices /path/to/russell_1000_prices.csv \ + --universe /path/to/russell_1000_universe.csv \ + --start 2018-01-01 \ + --end 2025-12-31 \ + --output-dir /path/to/backtest_outputs +``` + +输出目录默认会写： + +- `summary.csv` +- `portfolio_returns.csv` +- `weights_history.csv` +- `turnover_history.csv` + +**本地完整研究流程** + +1. 先把带日期的成分股快照目录整理成 interval 历史： + +```bash +PYTHONPATH=src:. python3 scripts/build_russell_1000_universe_history.py \ + --input-dir examples/russell_1000_universe_snapshots \ + --output /tmp/r1000_universe_history.csv +``` + +2. 再用 Yahoo Finance 拉价格历史： + +```bash +PYTHONPATH=src:. python3 scripts/fetch_russell_1000_price_history.py \ + --universe-history /tmp/r1000_universe_history.csv \ + --output /tmp/r1000_price_history.csv \ + --start 2024-01-01 +``` + +3. 生成最新 feature snapshot： + +```bash +PYTHONPATH=src:. python3 scripts/generate_russell_1000_feature_snapshot.py \ + --prices /tmp/r1000_price_history.csv \ + --universe /tmp/r1000_universe_history.csv \ + --output /tmp/r1000_feature_snapshot.csv +``` + +4. 最后跑回测： + +```bash +PYTHONPATH=src:. python3 scripts/backtest_russell_1000_multi_factor_defensive.py \ + --prices /tmp/r1000_price_history.csv \ + --universe /tmp/r1000_universe_history.csv \ + --output-dir /tmp/r1000_backtest +``` ### hybrid_growth_income diff --git a/src/us_equity_strategies/catalog.py b/src/us_equity_strategies/catalog.py index 22849a01..ca6740ef 100644 --- a/src/us_equity_strategies/catalog.py +++ b/src/us_equity_strategies/catalog.py @@ -9,6 +9,7 @@ GLOBAL_ETF_ROTATION_PROFILE = "global_etf_rotation" HYBRID_GROWTH_INCOME_PROFILE = "hybrid_growth_income" SEMICONDUCTOR_ROTATION_INCOME_PROFILE = "semiconductor_rotation_income" +RUSSELL_1000_MULTI_FACTOR_DEFENSIVE_PROFILE = "russell_1000_multi_factor_defensive" CASH_BUFFER_BRANCH_DEFAULT_PROFILE = "cash_buffer_branch_default" STRATEGY_DEFINITIONS: dict[str, StrategyDefinition] = { @@ -45,6 +46,17 @@ ), ), ), + RUSSELL_1000_MULTI_FACTOR_DEFENSIVE_PROFILE: StrategyDefinition( + profile=RUSSELL_1000_MULTI_FACTOR_DEFENSIVE_PROFILE, + domain=US_EQUITY_DOMAIN, + supported_platforms=frozenset({"ibkr"}), + components=( + StrategyComponentDefinition( + name="signal_logic", + module_path="us_equity_strategies.strategies.russell_1000_multi_factor_defensive", + ), + ), + ), CASH_BUFFER_BRANCH_DEFAULT_PROFILE: StrategyDefinition( profile=CASH_BUFFER_BRANCH_DEFAULT_PROFILE, domain=US_EQUITY_DOMAIN, diff --git a/src/us_equity_strategies/snapshots/__init__.py b/src/us_equity_strategies/snapshots/__init__.py index 95c63ca8..a5162f38 100644 --- a/src/us_equity_strategies/snapshots/__init__.py +++ b/src/us_equity_strategies/snapshots/__init__.py @@ -1,3 +1,13 @@ -"""Snapshot builders for research and runtime feature files.""" +from .russell_1000_multi_factor_defensive import ( + FEATURE_SNAPSHOT_COLUMNS, + build_feature_snapshot, + read_table, + write_table, +) -__all__ = [] +__all__ = [ + "FEATURE_SNAPSHOT_COLUMNS", + "build_feature_snapshot", + "read_table", + "write_table", +] diff --git a/src/us_equity_strategies/snapshots/russell_1000_multi_factor_defensive.py b/src/us_equity_strategies/snapshots/russell_1000_multi_factor_defensive.py new file mode 100644 index 00000000..77f7eec2 --- /dev/null +++ b/src/us_equity_strategies/snapshots/russell_1000_multi_factor_defensive.py @@ -0,0 +1,323 @@ +from __future__ import annotations + +import math +from collections.abc import Mapping +from pathlib import Path + +import pandas as pd + +PRICE_HISTORY_REQUIRED_COLUMNS = frozenset({"symbol", "as_of", "close", "volume"}) +UNIVERSE_REQUIRED_COLUMNS = frozenset({"symbol", "sector"}) +FEATURE_SNAPSHOT_COLUMNS = ( + "as_of", + "symbol", + "sector", + "close", + "volume", + "adv20_usd", + "history_days", + "mom_6_1", + "mom_12_1", + "sma200_gap", + "vol_63", + "maxdd_126", + "eligible", +) + + +def read_table(path: str | Path) -> pd.DataFrame: + raw_path = str(path or "").strip() + if not raw_path: + raise EnvironmentError("path is required") + table_path = Path(raw_path) + if not table_path.exists(): + raise FileNotFoundError(f"file not found: {table_path}") + + suffix = table_path.suffix.lower() + if suffix == ".csv": + return pd.read_csv(table_path) + if suffix in {".json", ".jsonl"}: + return pd.read_json(table_path, orient="records", lines=suffix == ".jsonl") + if suffix == ".parquet": + return pd.read_parquet(table_path) + raise ValueError("Unsupported table format; expected .csv, .json, .jsonl, or .parquet") + + +def write_table(frame: pd.DataFrame, path: str | Path) -> None: + raw_path = str(path or "").strip() + if not raw_path: + raise EnvironmentError("path is required") + table_path = Path(raw_path) + table_path.parent.mkdir(parents=True, exist_ok=True) + + suffix = table_path.suffix.lower() + if suffix == ".csv": + frame.to_csv(table_path, index=False) + return + if suffix == ".json": + frame.to_json(table_path, orient="records", indent=2, date_format="iso") + return + if suffix == ".jsonl": + frame.to_json(table_path, orient="records", lines=True, date_format="iso") + return + if suffix == ".parquet": + frame.to_parquet(table_path, index=False) + return + raise ValueError("Unsupported table format; expected .csv, .json, .jsonl, or .parquet") + + +def _require_columns(frame: pd.DataFrame, required: frozenset[str], *, name: str) -> None: + missing = required - set(frame.columns) + if missing: + missing_text = ", ".join(sorted(missing)) + raise ValueError(f"{name} missing required columns: {missing_text}") + + +def _normalize_symbol_series(values: pd.Series) -> pd.Series: + return values.astype(str).str.upper().str.strip() + + +def _normalize_date(value) -> pd.Timestamp: + timestamp = pd.Timestamp(value) + if timestamp.tzinfo is not None: + timestamp = timestamp.tz_convert(None) + return timestamp.normalize() + + +def _compute_skip_return(closes: pd.Series, *, skip_days: int, lookback_days: int) -> float: + required = skip_days + lookback_days + if len(closes) <= required: + return float("nan") + end_price = closes.iloc[-1 - skip_days] + start_price = closes.iloc[-1 - required] + if pd.isna(end_price) or pd.isna(start_price) or start_price <= 0: + return float("nan") + return float(end_price / start_price - 1.0) + + +def _compute_window_drawdown(closes: pd.Series) -> float: + if closes.empty: + return float("nan") + running_peak = closes.cummax() + drawdown = closes / running_peak - 1.0 + return float(drawdown.min()) + + +def _build_feature_row( + history: pd.DataFrame, + *, + symbol: str, + sector: str, + as_of: pd.Timestamp, + min_price_usd: float, + min_adv20_usd: float, + min_history_days: int, + momentum_skip_days: int, + momentum_6m_lookback_days: int, + momentum_12m_lookback_days: int, + sma_window: int, + vol_window: int, + drawdown_window: int, + force_ineligible: bool = False, +) -> dict[str, object]: + if history.empty: + return { + "as_of": as_of, + "symbol": symbol, + "sector": sector, + "close": float("nan"), + "volume": float("nan"), + "adv20_usd": float("nan"), + "history_days": 0, + "mom_6_1": float("nan"), + "mom_12_1": float("nan"), + "sma200_gap": float("nan"), + "vol_63": float("nan"), + "maxdd_126": float("nan"), + "eligible": False, + } + + history = history.sort_values("as_of") + closes = pd.to_numeric(history["close"], errors="coerce") + volumes = pd.to_numeric(history["volume"], errors="coerce") + dollar_volume = closes * volumes + returns = closes.pct_change() + + latest_close = float(closes.iloc[-1]) + latest_volume = float(volumes.iloc[-1]) if not pd.isna(volumes.iloc[-1]) else float("nan") + adv20_usd = float(dollar_volume.tail(20).mean()) if len(dollar_volume) >= 20 else float("nan") + mom_6_1 = _compute_skip_return( + closes, + skip_days=momentum_skip_days, + lookback_days=momentum_6m_lookback_days, + ) + mom_12_1 = _compute_skip_return( + closes, + skip_days=momentum_skip_days, + lookback_days=momentum_12m_lookback_days, + ) + sma200_gap = ( + float(latest_close / closes.tail(sma_window).mean() - 1.0) + if len(closes) >= sma_window + else float("nan") + ) + vol_63 = ( + float(returns.tail(vol_window).std(ddof=0) * math.sqrt(252)) + if returns.tail(vol_window).notna().sum() >= vol_window + else float("nan") + ) + maxdd_126 = ( + _compute_window_drawdown(closes.tail(drawdown_window)) + if len(closes) >= drawdown_window + else float("nan") + ) + + feature_values = (mom_6_1, mom_12_1, sma200_gap, vol_63, maxdd_126) + eligible = ( + not force_ineligible + and len(closes) >= min_history_days + and latest_close > min_price_usd + and not pd.isna(adv20_usd) + and adv20_usd >= min_adv20_usd + and all(not pd.isna(value) for value in feature_values) + ) + + return { + "as_of": as_of, + "symbol": symbol, + "sector": sector, + "close": latest_close, + "volume": latest_volume, + "adv20_usd": adv20_usd, + "history_days": int(len(closes)), + "mom_6_1": mom_6_1, + "mom_12_1": mom_12_1, + "sma200_gap": sma200_gap, + "vol_63": vol_63, + "maxdd_126": maxdd_126, + "eligible": bool(eligible), + } + + +def _normalize_price_groups( + price_history, + *, + as_of: pd.Timestamp, +) -> tuple[dict[str, pd.DataFrame], pd.DataFrame]: + if isinstance(price_history, Mapping): + price_groups: dict[str, pd.DataFrame] = {} + empty_history = pd.DataFrame(columns=["symbol", "as_of", "close", "volume"]) + for raw_symbol, raw_history in price_history.items(): + history = pd.DataFrame(raw_history).copy() + if history.empty: + continue + _require_columns(history, PRICE_HISTORY_REQUIRED_COLUMNS, name=f"price_history[{raw_symbol!r}]") + history["symbol"] = _normalize_symbol_series(history["symbol"]) + history["as_of"] = pd.to_datetime(history["as_of"], utc=False).map(_normalize_date) + history["close"] = pd.to_numeric(history["close"], errors="coerce") + history["volume"] = pd.to_numeric(history["volume"], errors="coerce") + history = history.dropna(subset=["symbol", "as_of", "close"]) + history = history.loc[history["as_of"] <= as_of].sort_values("as_of").reset_index(drop=True) + if history.empty: + continue + price_groups[str(raw_symbol).strip().upper()] = history + if empty_history.empty: + empty_history = history.iloc[0:0].copy() + return price_groups, empty_history + + prices = pd.DataFrame(price_history).copy() + if prices.empty: + raise ValueError("price_history must contain at least one row") + _require_columns(prices, PRICE_HISTORY_REQUIRED_COLUMNS, name="price_history") + + prices["symbol"] = _normalize_symbol_series(prices["symbol"]) + prices["as_of"] = pd.to_datetime(prices["as_of"], utc=False).map(_normalize_date) + prices["close"] = pd.to_numeric(prices["close"], errors="coerce") + prices["volume"] = pd.to_numeric(prices["volume"], errors="coerce") + prices = prices.dropna(subset=["symbol", "as_of", "close"]) + prices = prices.loc[prices["as_of"] <= as_of].copy() + price_groups = { + symbol: group.sort_values("as_of").reset_index(drop=True) + for symbol, group in prices.groupby("symbol", sort=False) + } + return price_groups, prices.iloc[0:0].copy() + + +def build_feature_snapshot( + price_history, + universe_snapshot, + *, + as_of_date=None, + benchmark_symbol: str = "SPY", + benchmark_sector: str = "benchmark", + min_price_usd: float = 10.0, + min_adv20_usd: float = 20_000_000.0, + min_history_days: int = 252, + momentum_skip_days: int = 21, + momentum_6m_lookback_days: int = 126, + momentum_12m_lookback_days: int = 252, + sma_window: int = 200, + vol_window: int = 63, + drawdown_window: int = 126, +) -> pd.DataFrame: + universe = pd.DataFrame(universe_snapshot).copy() + if universe.empty: + raise ValueError("universe_snapshot must contain at least one row") + _require_columns(universe, UNIVERSE_REQUIRED_COLUMNS, name="universe_snapshot") + + universe["symbol"] = _normalize_symbol_series(universe["symbol"]) + universe["sector"] = universe["sector"].fillna("unknown").astype(str).str.strip().replace("", "unknown") + universe = universe.drop_duplicates(subset=["symbol"], keep="last") + + if as_of_date is None: + if isinstance(price_history, Mapping): + max_dates = [] + for raw_history in price_history.values(): + history = pd.DataFrame(raw_history) + if "as_of" not in history.columns or history.empty: + continue + max_dates.append(pd.to_datetime(history["as_of"], utc=False).map(_normalize_date).max()) + if not max_dates: + raise ValueError("price_history must contain at least one usable row") + as_of = max(max_dates) + else: + prices = pd.DataFrame(price_history) + if prices.empty or "as_of" not in prices.columns: + raise ValueError("price_history must contain at least one usable row") + as_of = pd.to_datetime(prices["as_of"], utc=False).map(_normalize_date).max() + else: + as_of = _normalize_date(as_of_date) + price_groups, empty_history = _normalize_price_groups(price_history, as_of=as_of) + if not price_groups and empty_history.empty: + raise ValueError("price_history must contain at least one usable row") + + benchmark_symbol = str(benchmark_symbol or "").strip().upper() + symbols = universe["symbol"].tolist() + if benchmark_symbol and benchmark_symbol not in symbols: + symbols.append(benchmark_symbol) + + sector_map = dict(zip(universe["symbol"], universe["sector"])) + rows = [] + for symbol in symbols: + history = price_groups.get(symbol, empty_history) + rows.append( + _build_feature_row( + history, + symbol=symbol, + sector=sector_map.get(symbol, benchmark_sector if symbol == benchmark_symbol else "unknown"), + as_of=as_of, + min_price_usd=min_price_usd, + min_adv20_usd=min_adv20_usd, + min_history_days=min_history_days, + momentum_skip_days=momentum_skip_days, + momentum_6m_lookback_days=momentum_6m_lookback_days, + momentum_12m_lookback_days=momentum_12m_lookback_days, + sma_window=sma_window, + vol_window=vol_window, + drawdown_window=drawdown_window, + force_ineligible=symbol == benchmark_symbol, + ) + ) + + snapshot = pd.DataFrame(rows) + return snapshot.loc[:, FEATURE_SNAPSHOT_COLUMNS].sort_values(by=["symbol"]).reset_index(drop=True) diff --git a/src/us_equity_strategies/strategies/russell_1000_multi_factor_defensive.py b/src/us_equity_strategies/strategies/russell_1000_multi_factor_defensive.py new file mode 100644 index 00000000..e555c968 --- /dev/null +++ b/src/us_equity_strategies/strategies/russell_1000_multi_factor_defensive.py @@ -0,0 +1,361 @@ +from __future__ import annotations + +import math +from collections.abc import Mapping + +import pandas as pd + +SIGNAL_SOURCE = "feature_snapshot" +STATUS_ICON = "📏" +BENCHMARK_SYMBOL = "SPY" +SAFE_HAVEN = "BOXX" +DEFAULT_HOLDINGS_COUNT = 24 +DEFAULT_SINGLE_NAME_CAP = 0.06 +DEFAULT_SECTOR_CAP = 0.20 +DEFAULT_HOLD_BONUS = 0.15 +DEFAULT_SOFT_DEFENSE_EXPOSURE = 0.50 +DEFAULT_HARD_DEFENSE_EXPOSURE = 0.10 +DEFAULT_SOFT_BREADTH_THRESHOLD = 0.55 +DEFAULT_HARD_BREADTH_THRESHOLD = 0.35 + +REQUIRED_FEATURE_COLUMNS = frozenset( + { + "symbol", + "sector", + "mom_6_1", + "mom_12_1", + "sma200_gap", + "vol_63", + "maxdd_126", + } +) + + +def _coerce_bool(value) -> bool: + if pd.isna(value): + return False + if isinstance(value, bool): + return value + if isinstance(value, (int, float)): + return bool(value) + normalized = str(value).strip().lower() + if normalized in {"1", "true", "yes", "y"}: + return True + if normalized in {"0", "false", "no", "n"}: + return False + return bool(normalized) + + +def _normalize_holdings(current_holdings) -> set[str]: + if current_holdings is None: + return set() + if isinstance(current_holdings, Mapping): + raw_symbols = current_holdings.keys() + else: + raw_symbols = current_holdings + + normalized: set[str] = set() + for item in raw_symbols: + symbol = getattr(item, "symbol", item) + symbol_text = str(symbol or "").strip().upper() + if symbol_text: + normalized.add(symbol_text) + return normalized + + +def _zscore(values: pd.Series) -> pd.Series: + numeric = pd.to_numeric(values, errors="coerce") + std = numeric.std(ddof=0) + if pd.isna(std) or std == 0: + return pd.Series(0.0, index=values.index, dtype=float) + return ((numeric - numeric.mean()) / std).fillna(0.0) + + +def _to_frame(feature_snapshot) -> pd.DataFrame: + if isinstance(feature_snapshot, pd.DataFrame): + frame = feature_snapshot.copy() + else: + frame = pd.DataFrame(list(feature_snapshot)) + + if frame.empty: + raise ValueError("feature_snapshot must contain at least one row") + + missing = REQUIRED_FEATURE_COLUMNS - set(frame.columns) + if missing: + missing_text = ", ".join(sorted(missing)) + raise ValueError(f"feature_snapshot missing required columns: {missing_text}") + + frame["symbol"] = frame["symbol"].astype(str).str.upper().str.strip() + frame["sector"] = frame["sector"].fillna("unknown").astype(str).str.strip().replace("", "unknown") + if "eligible" in frame.columns: + frame["eligible"] = frame["eligible"].where(frame["eligible"].notna(), True) + else: + frame["eligible"] = True + frame["eligible"] = frame["eligible"].map(_coerce_bool) + + for column in ("mom_6_1", "mom_12_1", "sma200_gap", "vol_63", "maxdd_126"): + frame[column] = pd.to_numeric(frame[column], errors="coerce") + + return frame + + +def _resolve_regime( + *, + benchmark_trend_positive: bool, + breadth_ratio: float, + soft_breadth_threshold: float, + hard_breadth_threshold: float, +) -> str: + if (not benchmark_trend_positive) and breadth_ratio < hard_breadth_threshold: + return "hard_defense" + if (not benchmark_trend_positive) or breadth_ratio < soft_breadth_threshold: + return "soft_defense" + return "risk_on" + + +def _stock_exposure_for_regime( + regime: str, + *, + soft_defense_exposure: float, + hard_defense_exposure: float, +) -> float: + if regime == "hard_defense": + return float(hard_defense_exposure) + if regime == "soft_defense": + return float(soft_defense_exposure) + return 1.0 + + +def _select_symbols( + ranked: pd.DataFrame, + *, + holdings_count: int, + sector_slot_cap: int, +) -> pd.DataFrame: + selected_rows = [] + sector_counts: dict[str, int] = {} + for row in ranked.itertuples(index=False): + sector = row.sector + if sector_counts.get(sector, 0) >= sector_slot_cap: + continue + selected_rows.append(row._asdict()) + sector_counts[sector] = sector_counts.get(sector, 0) + 1 + if len(selected_rows) >= holdings_count: + break + + return pd.DataFrame(selected_rows) + + +def build_target_weights( + feature_snapshot, + current_holdings, + *, + benchmark_symbol: str = BENCHMARK_SYMBOL, + safe_haven: str = SAFE_HAVEN, + holdings_count: int = DEFAULT_HOLDINGS_COUNT, + single_name_cap: float = DEFAULT_SINGLE_NAME_CAP, + sector_cap: float = DEFAULT_SECTOR_CAP, + hold_bonus: float = DEFAULT_HOLD_BONUS, + soft_defense_exposure: float = DEFAULT_SOFT_DEFENSE_EXPOSURE, + hard_defense_exposure: float = DEFAULT_HARD_DEFENSE_EXPOSURE, + soft_breadth_threshold: float = DEFAULT_SOFT_BREADTH_THRESHOLD, + hard_breadth_threshold: float = DEFAULT_HARD_BREADTH_THRESHOLD, +): + """ + Build a price-only Russell 1000 target-weight plan from a precomputed feature snapshot. + + Expected feature columns: + - symbol + - sector + - mom_6_1 + - mom_12_1 + - sma200_gap + - vol_63 + - maxdd_126 + - eligible (optional, defaults to True) + + The benchmark row (default `SPY`) can be included in the same snapshot. + If present, its `sma200_gap` drives the market-regime filter. + """ + if holdings_count <= 0: + raise ValueError("holdings_count must be positive") + if single_name_cap <= 0 or sector_cap <= 0: + raise ValueError("single_name_cap and sector_cap must be positive") + + frame = _to_frame(feature_snapshot) + benchmark_symbol = str(benchmark_symbol or "").strip().upper() + safe_haven = str(safe_haven or "").strip().upper() + current_holdings_set = _normalize_holdings(current_holdings) + + benchmark_rows = frame.loc[frame["symbol"] == benchmark_symbol] + benchmark_trend_positive = True + if not benchmark_rows.empty: + benchmark_trend_positive = bool(benchmark_rows.iloc[-1]["sma200_gap"] > 0) + + universe = frame.loc[ + (frame["symbol"] != benchmark_symbol) & (frame["symbol"] != safe_haven) + ].copy() + eligible = universe.loc[ + universe["eligible"] + & universe["mom_6_1"].notna() + & universe["mom_12_1"].notna() + & universe["sma200_gap"].notna() + & universe["vol_63"].notna() + & universe["maxdd_126"].notna() + ].copy() + + breadth_ratio = float((eligible["sma200_gap"] > 0).mean()) if not eligible.empty else 0.0 + regime = _resolve_regime( + benchmark_trend_positive=benchmark_trend_positive, + breadth_ratio=breadth_ratio, + soft_breadth_threshold=soft_breadth_threshold, + hard_breadth_threshold=hard_breadth_threshold, + ) + stock_exposure = _stock_exposure_for_regime( + regime, + soft_defense_exposure=soft_defense_exposure, + hard_defense_exposure=hard_defense_exposure, + ) + + if eligible.empty or stock_exposure <= 0: + signal = ( + f"regime={regime} breadth={breadth_ratio:.1%} " + f"benchmark_trend={'up' if benchmark_trend_positive else 'down'}" + ) + return {safe_haven: 1.0}, signal, { + "benchmark_symbol": benchmark_symbol, + "benchmark_trend_positive": benchmark_trend_positive, + "breadth_ratio": breadth_ratio, + "regime": regime, + "stock_exposure": 0.0, + "selected_symbols": (), + "candidate_count": int(len(eligible)), + } + + eligible["z_mom_6_1"] = eligible.groupby("sector")["mom_6_1"].transform(_zscore) + eligible["z_mom_12_1"] = eligible.groupby("sector")["mom_12_1"].transform(_zscore) + eligible["z_sma200_gap"] = eligible.groupby("sector")["sma200_gap"].transform(_zscore) + eligible["z_vol_63"] = eligible.groupby("sector")["vol_63"].transform(_zscore) + eligible["drawdown_abs"] = eligible["maxdd_126"].abs() + eligible["z_drawdown_abs"] = eligible.groupby("sector")["drawdown_abs"].transform(_zscore) + eligible["score"] = ( + (eligible["z_mom_6_1"] * 0.35) + + (eligible["z_mom_12_1"] * 0.30) + + (eligible["z_sma200_gap"] * 0.15) + - (eligible["z_vol_63"] * 0.10) + - (eligible["z_drawdown_abs"] * 0.10) + ) + eligible.loc[ + eligible["symbol"].isin(current_holdings_set), + "score", + ] += float(hold_bonus) + + ranked = eligible.sort_values( + by=["score", "mom_12_1", "mom_6_1", "symbol"], + ascending=[False, False, False, True], + ) + + per_name_target = stock_exposure / holdings_count + if per_name_target <= 0: + sector_slot_cap = holdings_count + else: + sector_slot_cap = max(1, int(math.floor(sector_cap / per_name_target))) + + selected = _select_symbols( + ranked, + holdings_count=holdings_count, + sector_slot_cap=sector_slot_cap, + ) + + if selected.empty: + signal = ( + f"regime={regime} breadth={breadth_ratio:.1%} " + f"benchmark_trend={'up' if benchmark_trend_positive else 'down'} no_selection" + ) + return {safe_haven: 1.0}, signal, { + "benchmark_symbol": benchmark_symbol, + "benchmark_trend_positive": benchmark_trend_positive, + "breadth_ratio": breadth_ratio, + "regime": regime, + "stock_exposure": 0.0, + "selected_symbols": (), + "candidate_count": int(len(eligible)), + } + + per_name_weight = min(single_name_cap, stock_exposure / len(selected)) + invested_weight = per_name_weight * len(selected) + weights = {row.symbol: per_name_weight for row in selected.itertuples(index=False)} + if invested_weight < 1.0: + weights[safe_haven] = 1.0 - invested_weight + + top_preview = ", ".join( + f"{row.symbol}({row.score:.2f})" + for row in selected.head(5).itertuples(index=False) + ) + signal = ( + f"regime={regime} breadth={breadth_ratio:.1%} " + f"benchmark_trend={'up' if benchmark_trend_positive else 'down'} " + f"stock_exposure={stock_exposure:.1%} selected={len(selected)} top={top_preview}" + ) + metadata = { + "benchmark_symbol": benchmark_symbol, + "benchmark_trend_positive": benchmark_trend_positive, + "breadth_ratio": breadth_ratio, + "regime": regime, + "stock_exposure": stock_exposure, + "selected_symbols": tuple(selected["symbol"].tolist()), + "candidate_count": int(len(eligible)), + "sector_slot_cap": sector_slot_cap, + } + return weights, signal, metadata + + +def extract_managed_symbols( + feature_snapshot, + *, + benchmark_symbol: str = BENCHMARK_SYMBOL, + safe_haven: str = SAFE_HAVEN, +) -> tuple[str, ...]: + frame = _to_frame(feature_snapshot) + benchmark_symbol = str(benchmark_symbol or "").strip().upper() + safe_haven = str(safe_haven or "").strip().upper() + + symbols = [] + for symbol in frame["symbol"].tolist(): + if symbol == benchmark_symbol: + continue + symbols.append(symbol) + if safe_haven and safe_haven not in symbols: + symbols.append(safe_haven) + return tuple(dict.fromkeys(symbols)) + + +def compute_signals(feature_snapshot, current_holdings, **kwargs): + weights, signal_desc, metadata = build_target_weights( + feature_snapshot, + current_holdings, + **kwargs, + ) + benchmark_symbol = kwargs.get("benchmark_symbol", BENCHMARK_SYMBOL) + safe_haven = kwargs.get("safe_haven", SAFE_HAVEN) + managed_symbols = extract_managed_symbols( + feature_snapshot, + benchmark_symbol=benchmark_symbol, + safe_haven=safe_haven, + ) + status_desc = ( + f"breadth={metadata['breadth_ratio']:.1%} | " + f"regime={metadata['regime']} | " + f"benchmark={'up' if metadata['benchmark_trend_positive'] else 'down'}" + ) + return ( + weights, + signal_desc, + metadata["regime"] == "hard_defense", + status_desc, + { + **metadata, + "managed_symbols": managed_symbols, + "status_icon": STATUS_ICON, + }, + ) From 5c40d49927b7a86e27feee23b82da81338320404 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 5 Apr 2026 05:00:05 +0800 Subject: [PATCH 3/3] fix ruff lint --- .../strategies/cash_buffer_branch_default.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/src/us_equity_strategies/strategies/cash_buffer_branch_default.py b/src/us_equity_strategies/strategies/cash_buffer_branch_default.py index a47ed623..d0cb135f 100644 --- a/src/us_equity_strategies/strategies/cash_buffer_branch_default.py +++ b/src/us_equity_strategies/strategies/cash_buffer_branch_default.py @@ -594,13 +594,17 @@ def evaluate_execution_window( } +def _noop_logger(_message) -> None: + return None + + def load_runtime_parameters( *, config_path: str | Path | None = None, logger=None, ) -> dict[str, object]: if logger is None: - logger = lambda message: None + logger = _noop_logger runtime_params = { "benchmark_symbol": BENCHMARK_SYMBOL,