diff --git a/src/hk_equity_strategies/catalog.py b/src/hk_equity_strategies/catalog.py index a4cd49b..1122ecc 100644 --- a/src/hk_equity_strategies/catalog.py +++ b/src/hk_equity_strategies/catalog.py @@ -18,9 +18,12 @@ from hk_equity_strategies.strategies import hk_global_etf_tactical_rotation as global_etf_strategy from hk_equity_strategies.strategies import hk_low_vol_dividend_quality_snapshot as low_vol_dividend_strategy +from hk_equity_strategies.strategies import hk_equity_combo as hk_combo_strategy + HK_EQUITY_DOMAIN = global_etf_strategy.HK_EQUITY_DOMAIN HK_GLOBAL_ETF_TACTICAL_ROTATION_PROFILE = global_etf_strategy.PROFILE_NAME HK_LOW_VOL_DIVIDEND_QUALITY_SNAPSHOT_PROFILE = low_vol_dividend_strategy.PROFILE_NAME +HK_EQUITY_COMBO_PROFILE = hk_combo_strategy.PROFILE_NAME HK_DIRECT_MARKET_HISTORY_PROFILES = frozenset( { @@ -34,11 +37,13 @@ STRATEGY_PLATFORM_COMPATIBILITY: dict[str, frozenset[str]] = { HK_GLOBAL_ETF_TACTICAL_ROTATION_PROFILE: frozenset({"ibkr", "longbridge"}), HK_LOW_VOL_DIVIDEND_QUALITY_SNAPSHOT_PROFILE: frozenset({"ibkr", "longbridge"}), + HK_EQUITY_COMBO_PROFILE: frozenset({"manual"}), } STRATEGY_REQUIRED_INPUTS: dict[str, frozenset[str]] = { HK_GLOBAL_ETF_TACTICAL_ROTATION_PROFILE: frozenset({"market_history"}), HK_LOW_VOL_DIVIDEND_QUALITY_SNAPSHOT_PROFILE: frozenset({"feature_snapshot"}), + HK_EQUITY_COMBO_PROFILE: frozenset({"market_history", "dividend_snapshot"}), } STRATEGY_DEFAULT_CONFIG: dict[str, dict[str, object]] = { @@ -78,16 +83,24 @@ "execution_cash_reserve_ratio": low_vol_dividend_strategy.DEFAULT_EXECUTION_CASH_RESERVE_RATIO, "rebalance_frequency": "monthly", }, + HK_EQUITY_COMBO_PROFILE: { + "etf_weight": 0.60, + "dividend_weight": 0.40, + "execution_cash_reserve_ratio": 0.02, + "rebalance_frequency": "monthly", + }, } STRATEGY_ENTRYPOINT_ATTRIBUTES: dict[str, str] = { HK_GLOBAL_ETF_TACTICAL_ROTATION_PROFILE: "hk_global_etf_tactical_rotation_entrypoint", HK_LOW_VOL_DIVIDEND_QUALITY_SNAPSHOT_PROFILE: "hk_low_vol_dividend_quality_snapshot_entrypoint", + HK_EQUITY_COMBO_PROFILE: "hk_equity_combo_entrypoint", } STRATEGY_TARGET_MODES: dict[str, str] = { HK_GLOBAL_ETF_TACTICAL_ROTATION_PROFILE: "weight", HK_LOW_VOL_DIVIDEND_QUALITY_SNAPSHOT_PROFILE: "weight", + HK_EQUITY_COMBO_PROFILE: "weight", } @@ -130,6 +143,11 @@ def _build_strategy_definition( component_name="signal_logic", module_path="hk_equity_strategies.strategies.hk_low_vol_dividend_quality_snapshot", ), + HK_EQUITY_COMBO_PROFILE: _build_strategy_definition( + HK_EQUITY_COMBO_PROFILE, + component_name="signal_logic", + module_path="hk_equity_strategies.strategies.hk_equity_combo", + ), } STRATEGY_METADATA: dict[str, StrategyMetadata] = { @@ -163,6 +181,17 @@ def _build_strategy_definition( role="hk_snapshot_low_vol_dividend_quality", status="runtime_enabled", ), + HK_EQUITY_COMBO_PROFILE: StrategyMetadata( + canonical_profile=HK_EQUITY_COMBO_PROFILE, + display_name="HK Equity Combo", + description="Combined HK equity strategy: Global ETF tactical rotation (60%) + low-vol dividend quality (40%) blended portfolio.", + aliases=(), + cadence="monthly review", + asset_scope="hk_equity_combo", + benchmark="2800", + role="hk_equity_combo", + status="runtime_enabled", + ), } PROFILE_ALIASES: dict[str, str] = { diff --git a/src/hk_equity_strategies/combo_entrypoints.py b/src/hk_equity_strategies/combo_entrypoints.py new file mode 100644 index 0000000..11cde58 --- /dev/null +++ b/src/hk_equity_strategies/combo_entrypoints.py @@ -0,0 +1,48 @@ +"""HK equity combo entrypoints — moved from QuantHkComboStrategies. +""" +from __future__ import annotations + +from typing import Any + +from quant_platform_kit.strategy_contracts import ( + CallableStrategyEntrypoint, PositionTarget, StrategyContext, StrategyDecision, +) + +from hk_equity_strategies.combo_manifests import hk_equity_combo_manifest +from hk_equity_strategies.strategies import hk_equity_combo + + +def _require_market_data(ctx: StrategyContext, key: str) -> Any: + if key not in ctx.market_data: + raise ValueError(f"StrategyContext.market_data[{key!r}] is required") + return ctx.market_data[key] + + +def evaluate_hk_equity_combo(ctx: StrategyContext) -> StrategyDecision: + config = {**hk_equity_combo_manifest.default_config, **ctx.runtime_config or {}} + config.pop("execution_cash_reserve_ratio", None) + config.pop("rebalance_frequency", None) + combined, metadata = hk_equity_combo.build_target_weights( + market_history=_require_market_data(ctx, "market_history"), + dividend_snapshot=_require_market_data(ctx, "dividend_snapshot"), + config=config, + ) + diagnostics = { + **metadata, "signal_description": f"etf={config.get('etf_weight', 0.60):.0%} div={config.get('dividend_weight', 0.40):.0%}", + "status_description": f"etf={config.get('etf_weight', 0.60):.0%} div={config.get('dividend_weight', 0.40):.0%}", + "signal_source": hk_equity_combo.SIGNAL_SOURCE, "actionable": True, + } + return StrategyDecision( + positions=tuple( + PositionTarget(symbol=str(s), target_weight=float(w), role="target") + for s, w in sorted(combined.items()) if abs(float(w)) > 1e-12 + ), + risk_flags=(), diagnostics=diagnostics, + ) + + +hk_equity_combo_entrypoint = CallableStrategyEntrypoint( + manifest=hk_equity_combo_manifest, _evaluate=evaluate_hk_equity_combo, +) + +__all__ = ["evaluate_hk_equity_combo", "hk_equity_combo_entrypoint"] diff --git a/src/hk_equity_strategies/combo_manifests.py b/src/hk_equity_strategies/combo_manifests.py new file mode 100644 index 0000000..abe90c0 --- /dev/null +++ b/src/hk_equity_strategies/combo_manifests.py @@ -0,0 +1,33 @@ +"""HK equity combo manifests — moved from QuantHkComboStrategies. +""" +from __future__ import annotations + +from quant_platform_kit.strategy_contracts import StrategyManifest + +from hk_equity_strategies.strategies import hk_equity_combo as combo_strategy + +HK_EQUITY_COMBO_PROFILE = combo_strategy.PROFILE_NAME + + +def _manifest(*, profile, domain, display_name, description, aliases=(), required_inputs=frozenset(), default_config=None): + return StrategyManifest( + profile=profile, domain=domain, display_name=display_name, + description=description, aliases=aliases, + required_inputs=required_inputs, default_config=default_config or {}, + ) + + +hk_equity_combo_manifest = _manifest( + profile=HK_EQUITY_COMBO_PROFILE, + domain="hk_equity", + display_name="HK Equity Combo", + description="Combined HK equity strategy: Global ETF tactical rotation (60%) + low-vol dividend quality (40%) blended portfolio.", + aliases=(), + required_inputs=frozenset({"market_history", "dividend_snapshot"}), + default_config={ + "etf_weight": 0.60, "dividend_weight": 0.40, + "execution_cash_reserve_ratio": 0.02, "rebalance_frequency": "monthly", + }, +) + +__all__ = ["HK_EQUITY_COMBO_PROFILE", "hk_equity_combo_manifest"] diff --git a/src/hk_equity_strategies/entrypoints/__init__.py b/src/hk_equity_strategies/entrypoints/__init__.py index b34bfcf..14f01b0 100644 --- a/src/hk_equity_strategies/entrypoints/__init__.py +++ b/src/hk_equity_strategies/entrypoints/__init__.py @@ -76,9 +76,29 @@ def evaluate_hk_low_vol_dividend_quality_snapshot(ctx: StrategyContext) -> Strat ) +# --------------------------------------------------------------------------- +# HK Equity Combo entrypoint — delegates to combo_entrypoints +# --------------------------------------------------------------------------- + + +def evaluate_hk_equity_combo(ctx: StrategyContext) -> StrategyDecision: + from hk_equity_strategies.combo_entrypoints import evaluate_hk_equity_combo as _eval + return _eval(ctx) + + +from hk_equity_strategies.combo_manifests import hk_equity_combo_manifest # noqa: E402 — intentional late import + +hk_equity_combo_entrypoint = CallableStrategyEntrypoint( + manifest=hk_equity_combo_manifest, + _evaluate=evaluate_hk_equity_combo, +) + + __all__ = [ "evaluate_hk_global_etf_tactical_rotation", "evaluate_hk_low_vol_dividend_quality_snapshot", + "evaluate_hk_equity_combo", "hk_global_etf_tactical_rotation_entrypoint", "hk_low_vol_dividend_quality_snapshot_entrypoint", + "hk_equity_combo_entrypoint", ] diff --git a/src/hk_equity_strategies/strategies/hk_equity_combo.py b/src/hk_equity_strategies/strategies/hk_equity_combo.py new file mode 100644 index 0000000..b6d0b8b --- /dev/null +++ b/src/hk_equity_strategies/strategies/hk_equity_combo.py @@ -0,0 +1,205 @@ +"""HK equity combo strategy — 60/40 blend of ETF momentum and dividend-quality.""" + +from __future__ import annotations + +from typing import Any + +from quant_platform_kit.common.strategies import compute_portfolio_drift + +# HK sub-strategies +from hk_equity_strategies.strategies import ( + hk_global_etf_tactical_rotation as _etf, + hk_low_vol_dividend_quality_snapshot as _dividend, +) + +PROFILE_NAME = "hk_equity_combo" +SIGNAL_SOURCE = "combo" +STATUS_ICON = "\U0001f1ed\U0001f1f0" # Hong Kong flag + +DEFAULT_ETF_WEIGHT = 0.60 +DEFAULT_DIVIDEND_WEIGHT = 0.40 +DEFAULT_REBALANCE_THRESHOLD = 0.05 # 5% drift triggers rebalance + + +def _apply_dividend_regime( + etf_weight: float, + dividend_regime: str | None, +) -> tuple[float, float, str | None]: + """Apply a dividend-regime override to derive effective ETF / dividend weights. + + Parameters + ---------- + etf_weight : float + Configured ETF weight before regime adjustment. + dividend_regime : str | None + Regime indicator from runtime config. ``None`` or ``"risk_on"`` leaves + weights unchanged. + + Returns + ------- + tuple[float, float, str | None] + ``(effective_etf_weight, effective_dividend_weight, label)`` where *label* + is a human-readable regime name (or ``None`` when no adjustment applies). + """ + if dividend_regime is None or str(dividend_regime).strip().lower() == "risk_on": + return etf_weight, 1.0 - etf_weight, None + + regime = str(dividend_regime).strip().lower() + + if regime == "soft_defense": + effective_etf = min(etf_weight * 0.85, 1.0) + elif regime == "hard_defense": + effective_etf = min(etf_weight * 0.50, 1.0) + else: + return etf_weight, 1.0 - etf_weight, None + + return effective_etf, 1.0 - effective_etf, regime + + +def build_target_weights( + market_history: Any, + dividend_snapshot: Any, + config: dict[str, Any] | None = None, + **kwargs: Any, +) -> tuple[dict[str, float], dict[str, Any]]: + """Build a blended target-weight map by combining ETF momentum and dividend- + quality sub-strategies at the configured ratio. + + Parameters + ---------- + market_history : Any + Market price / volume data passed through to the ETF momentum sub-strategy. + dividend_snapshot : Any + Factor snapshot passed through to the dividend-quality sub-strategy. + config : dict[str, Any] | None + Optional override dictionary. Recognised keys: + + * ``etf_weight`` / ``dividend_weight`` (float, default 0.60 / 0.40) + * ``dividend_regime`` (str | None) — dynamic regime override + (``"risk_on"``, ``"soft_defense"``, or ``"hard_defense"``). + When set, the ETF weight is scaled down: ``risk_on`` = normal (no + change), ``soft_defense`` = ETF * 0.85, ``hard_defense`` = ETF * 0.50. + The freed allocation shifts to the dividend leg. + * ``etf_kwargs`` — forwarded to the ETF sub-strategy + * ``dividend_kwargs`` — forwarded to the dividend sub-strategy + + **kwargs + Additional keyword arguments forwarded to both sub-strategies. + + Returns + ------- + tuple[dict[str, float], dict[str, object]] + (combined_weight_map, metadata_dict). + """ + if config is None: + config = {} + + raw_etf_weight = config.get("etf_weight", DEFAULT_ETF_WEIGHT) + raw_dividend_weight = config.get("dividend_weight", DEFAULT_DIVIDEND_WEIGHT) + + etf_weight, dividend_weight, regime_label = _apply_dividend_regime( + raw_etf_weight, + config.get("dividend_regime"), + ) + + etf_kwargs: dict[str, Any] = {**kwargs, **config.get("etf_kwargs", {})} + dividend_kwargs: dict[str, Any] = {**kwargs, **config.get("dividend_kwargs", {})} + + etf_weights, etf_meta = _etf.build_target_weights(market_history, **etf_kwargs) + div_weights, div_df, div_meta = _dividend.build_target_weights( + dividend_snapshot, **dividend_kwargs + ) + + combined: dict[str, float] = {} + for sym, w in etf_weights.items(): + combined[sym] = combined.get(sym, 0.0) + w * etf_weight + for sym, w in div_weights.items(): + combined[sym] = combined.get(sym, 0.0) + w * dividend_weight + + metadata: dict[str, Any] = { + "etf_meta": etf_meta, + "dividend_meta": div_meta, + "etf_weight": etf_weight, + "dividend_weight": dividend_weight, + "raw_etf_weight": raw_etf_weight, + "raw_dividend_weight": raw_dividend_weight, + "dividend_regime": regime_label, + "profile": PROFILE_NAME, + "rebalance": compute_portfolio_drift( + combined, + holdings=config.get("current_holdings_quantities", {}), + prices=config.get("current_prices", {}), + threshold=float(config.get("rebalance_threshold", DEFAULT_REBALANCE_THRESHOLD)), + ), + } + + return combined, metadata + + +# _check_drift removed — use quant_platform_kit.common.strategies.compute_portfolio_drift + + +def compute_signals( + market_history: Any, + dividend_snapshot: Any, + config: dict[str, Any] | None = None, + **kwargs: Any, +) -> Any: + """Compute combined signals from both sub-strategies and produce a unified + signal frame / dict. + + Parameters + ---------- + market_history : Any + Market data forwarded to the ETF momentum sub-strategy. + dividend_snapshot : Any + Factor snapshot forwarded to the dividend-quality sub-strategy. + config : dict[str, Any] | None + Optional override dictionary (same layout as + :func:`build_target_weights`). + **kwargs + Additional keyword arguments forwarded to both sub-strategies. + + Returns + ------- + Any + Combines the sub-strategies' signal data into a single result. + """ + if config is None: + config = {} + + etf_kwargs: dict[str, Any] = {**kwargs, **config.get("etf_kwargs", {})} + dividend_kwargs: dict[str, Any] = {**kwargs, **config.get("dividend_kwargs", {})} + + etf_signals = _etf.compute_signals(market_history, **etf_kwargs) + div_signals = _dividend.compute_signals( + dividend_snapshot, current_holdings=None, **dividend_kwargs + ) + + return {"etf": etf_signals, "dividend": div_signals, "profile": PROFILE_NAME} + + +def extract_managed_symbols( + market_history: Any, + dividend_snapshot: Any, + **kwargs: Any, +) -> tuple[str, ...]: + """Return the union of symbols managed by both sub-strategies. + + Parameters + ---------- + market_history : Any + Market data forwarded to the ETF sub-strategy. + dividend_snapshot : Any + Factor snapshot forwarded to the dividend sub-strategy. + **kwargs + Additional keyword arguments forwarded to both sub-strategies. + + Returns + ------- + tuple[str, ...] + Deduplicated tuple of managed ticker symbols. + """ + etf_symbols = set(_etf.extract_managed_symbols(**kwargs)) + div_symbols = set(_dividend.extract_managed_symbols(dividend_snapshot, **kwargs)) + return tuple(sorted(etf_symbols | div_symbols))