diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 80a405a..a6d779c 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -12,6 +12,40 @@ jobs: - name: Checkout uses: actions/checkout@v6 + - name: Resolve QuantPlatformKit ref + id: quant-platform-kit-ref + run: | + set -euo pipefail + ref="main" + if [ -n "${GITHUB_HEAD_REF:-}" ] && git ls-remote --exit-code --heads https://github.com/QuantStrategyLab/QuantPlatformKit.git "${GITHUB_HEAD_REF}" >/dev/null 2>&1; then + ref="${GITHUB_HEAD_REF}" + fi + echo "ref=${ref}" >> "$GITHUB_OUTPUT" + + - name: Resolve UsEquityStrategies ref + id: us-equity-strategies-ref + run: | + set -euo pipefail + ref="main" + if [ -n "${GITHUB_HEAD_REF:-}" ] && git ls-remote --exit-code --heads https://github.com/QuantStrategyLab/UsEquityStrategies.git "${GITHUB_HEAD_REF}" >/dev/null 2>&1; then + ref="${GITHUB_HEAD_REF}" + fi + echo "ref=${ref}" >> "$GITHUB_OUTPUT" + + - name: Checkout QuantPlatformKit + uses: actions/checkout@v6 + with: + repository: QuantStrategyLab/QuantPlatformKit + ref: ${{ steps.quant-platform-kit-ref.outputs.ref }} + path: external/QuantPlatformKit + + - name: Checkout UsEquityStrategies + uses: actions/checkout@v6 + with: + repository: QuantStrategyLab/UsEquityStrategies + ref: ${{ steps.us-equity-strategies-ref.outputs.ref }} + path: external/UsEquityStrategies + - name: Setup Python uses: actions/setup-python@v6 with: @@ -22,6 +56,7 @@ jobs: set -euo pipefail python -m pip install --upgrade pip python -m pip install -r requirements.txt pytest ruff + python -m pip install --no-deps -e external/QuantPlatformKit -e external/UsEquityStrategies - name: Run ruff run: | @@ -31,4 +66,4 @@ jobs: - name: Run unit tests run: | set -euo pipefail - PYTHONPATH=. PYTEST_DISABLE_PLUGIN_AUTOLOAD=1 python -m pytest -q + PYTHONPATH=. PYTEST_DISABLE_PLUGIN_AUTOLOAD=1 python -m pytest -q tests diff --git a/decision_mapper.py b/decision_mapper.py new file mode 100644 index 0000000..9ea0f84 --- /dev/null +++ b/decision_mapper.py @@ -0,0 +1,110 @@ +from __future__ import annotations + +from collections.abc import Mapping +from typing import Any + +from quant_platform_kit.strategy_contracts import StrategyDecision + + +_EMERGENCY_FLAGS = frozenset({"emergency", "hard_defense"}) +_NO_EXECUTE_FLAGS = frozenset({"no_execute"}) + + +def _derive_target_weights(decision: StrategyDecision) -> dict[str, float]: + weights: dict[str, float] = {} + for position in decision.positions: + if position.target_weight is None: + raise ValueError( + "IBKR decision mapper only supports weight-based positions; " + f"position {position.symbol!r} is missing target_weight" + ) + weights[position.symbol] = float(position.target_weight) + return weights + + +def _derive_managed_symbols( + decision: StrategyDecision, + runtime_metadata: Mapping[str, Any], +) -> tuple[str, ...]: + explicit = runtime_metadata.get("managed_symbols") + if explicit: + return tuple(str(symbol) for symbol in explicit) + return tuple(position.symbol for position in decision.positions) + + +def _derive_safe_haven_symbol( + decision: StrategyDecision, + runtime_metadata: Mapping[str, Any], +) -> str | None: + explicit = runtime_metadata.get("safe_haven_symbol") + if explicit: + return str(explicit) + for position in decision.positions: + if position.role == "safe_haven": + return position.symbol + return None + + +def _derive_signal_description( + decision: StrategyDecision, + runtime_metadata: Mapping[str, Any], +) -> str: + diagnostics = decision.diagnostics + candidates = ( + diagnostics.get("signal_description"), + diagnostics.get("signal_display"), + diagnostics.get("signal_message"), + runtime_metadata.get("signal_description"), + runtime_metadata.get("signal_display"), + ) + for candidate in candidates: + text = str(candidate or "").strip() + if text: + return text + return "decision_ready" + + +def _derive_status_description( + decision: StrategyDecision, + runtime_metadata: Mapping[str, Any], +) -> str: + diagnostics = decision.diagnostics + candidates = ( + diagnostics.get("status_description"), + diagnostics.get("canary_status"), + diagnostics.get("market_status"), + runtime_metadata.get("status_description"), + ) + for candidate in candidates: + text = str(candidate or "").strip() + if text: + return text + return _derive_signal_description(decision, runtime_metadata) + + +def map_strategy_decision( + decision: StrategyDecision, + *, + strategy_profile: str, + runtime_metadata: Mapping[str, Any] | None = None, +) -> tuple[dict[str, float] | None, str, bool, str, dict[str, Any]]: + runtime_metadata = dict(runtime_metadata or {}) + diagnostics = dict(decision.diagnostics) + risk_flags = tuple(str(flag) for flag in decision.risk_flags) + no_execute = bool(_NO_EXECUTE_FLAGS & set(risk_flags)) + target_weights = None if no_execute else _derive_target_weights(decision) + signal_desc = _derive_signal_description(decision, runtime_metadata) + status_desc = _derive_status_description(decision, runtime_metadata) + is_emergency = bool(_EMERGENCY_FLAGS & set(risk_flags)) + + metadata: dict[str, Any] = {**runtime_metadata, **diagnostics} + metadata.setdefault("strategy_profile", strategy_profile) + metadata.setdefault("status_icon", "🐤") + metadata.setdefault("managed_symbols", _derive_managed_symbols(decision, runtime_metadata)) + safe_haven_symbol = _derive_safe_haven_symbol(decision, runtime_metadata) + if safe_haven_symbol: + metadata.setdefault("safe_haven_symbol", safe_haven_symbol) + metadata.setdefault("risk_flags", risk_flags) + metadata.setdefault("actionable", not no_execute) + + return target_weights, signal_desc, is_emergency, status_desc, metadata diff --git a/main.py b/main.py index 9167057..55a41c7 100644 --- a/main.py +++ b/main.py @@ -29,14 +29,14 @@ execute_rebalance as application_execute_rebalance, get_market_prices as application_get_market_prices, ) -from application.feature_snapshot_service import load_feature_snapshot_guarded from application.rebalance_service import run_strategy_core as run_rebalance_cycle +from decision_mapper import map_strategy_decision from entrypoints.cloud_run import is_market_open_today from runtime_config_support import ( load_platform_runtime_settings, resolve_ib_gateway_ip_mode, ) -from strategy_loader import load_signal_logic_module +from strategy_runtime import load_strategy_runtime app = Flask(__name__) ensure_event_loop = ibkr_ensure_event_loop @@ -132,123 +132,44 @@ def get_ib_port(): SERVICE_NAME = RUNTIME_SETTINGS.service_name ACCOUNT_IDS = RUNTIME_SETTINGS.account_ids -STRATEGY_LOGIC = load_signal_logic_module(STRATEGY_PROFILE) -STRATEGY_SIGNAL_SOURCE = getattr(STRATEGY_LOGIC, "SIGNAL_SOURCE", "market_data") -STRATEGY_STATUS_ICON = getattr(STRATEGY_LOGIC, "STATUS_ICON", "🐤") -SAFE_HAVEN = getattr(STRATEGY_LOGIC, "SAFE_HAVEN", "BIL") -RANKING_POOL = list(getattr(STRATEGY_LOGIC, "RANKING_POOL", ())) -CANARY_ASSETS = list(getattr(STRATEGY_LOGIC, "CANARY_ASSETS", ())) -TOP_N = getattr(STRATEGY_LOGIC, "TOP_N", None) -SMA_PERIOD = getattr(STRATEGY_LOGIC, "SMA_PERIOD", 200) -CANARY_BAD_THRESHOLD = getattr(STRATEGY_LOGIC, "CANARY_BAD_THRESHOLD", None) -REBALANCE_MONTHS = getattr(STRATEGY_LOGIC, "REBALANCE_MONTHS", None) -FEATURE_SNAPSHOT_PATH = RUNTIME_SETTINGS.feature_snapshot_path -FEATURE_SNAPSHOT_MANIFEST_PATH = RUNTIME_SETTINGS.feature_snapshot_manifest_path -FEATURE_REQUIRE_SNAPSHOT_MANIFEST = bool(getattr(STRATEGY_LOGIC, "REQUIRE_SNAPSHOT_MANIFEST", False)) -FEATURE_SNAPSHOT_CONTRACT_VERSION = getattr(STRATEGY_LOGIC, "SNAPSHOT_CONTRACT_VERSION", None) -feature_runtime_loader = getattr(STRATEGY_LOGIC, "load_runtime_parameters", None) -FEATURE_RUNTIME_PARAMETERS = ( - feature_runtime_loader( - config_path=RUNTIME_SETTINGS.strategy_config_path, - logger=lambda message: print(message, flush=True), - ) - if STRATEGY_SIGNAL_SOURCE == "feature_snapshot" and callable(feature_runtime_loader) - else {} -) -HOLD_BONUS = FEATURE_RUNTIME_PARAMETERS.get( - "hold_bonus", - getattr(STRATEGY_LOGIC, "HOLD_BONUS", getattr(STRATEGY_LOGIC, "DEFAULT_HOLD_BONUS", 0.0)), -) -FEATURE_SIGNAL_KWARG_KEYS = tuple( - getattr( - STRATEGY_LOGIC, - "FEATURE_SIGNAL_KWARG_KEYS", - ( - "benchmark_symbol", - "safe_haven", - "holdings_count", - "single_name_cap", - "sector_cap", - "hold_bonus", - "soft_defense_exposure", - "hard_defense_exposure", - "soft_breadth_threshold", - "hard_breadth_threshold", - ), - ) -) -FEATURE_BENCHMARK_SYMBOL = FEATURE_RUNTIME_PARAMETERS.get( - "benchmark_symbol", - getattr(STRATEGY_LOGIC, "BENCHMARK_SYMBOL", "SPY"), -) -FEATURE_HOLDINGS_COUNT = FEATURE_RUNTIME_PARAMETERS.get( - "holdings_count", - getattr(STRATEGY_LOGIC, "DEFAULT_HOLDINGS_COUNT", 24), -) -FEATURE_SINGLE_NAME_CAP = FEATURE_RUNTIME_PARAMETERS.get( - "single_name_cap", - getattr(STRATEGY_LOGIC, "DEFAULT_SINGLE_NAME_CAP", 0.06), -) -FEATURE_SECTOR_CAP = FEATURE_RUNTIME_PARAMETERS.get( - "sector_cap", - getattr(STRATEGY_LOGIC, "DEFAULT_SECTOR_CAP", 0.20), -) -FEATURE_RISK_ON_EXPOSURE = FEATURE_RUNTIME_PARAMETERS.get( - "risk_on_exposure", - 1.0, -) -FEATURE_SOFT_DEFENSE_EXPOSURE = FEATURE_RUNTIME_PARAMETERS.get( - "soft_defense_exposure", - getattr(STRATEGY_LOGIC, "DEFAULT_SOFT_DEFENSE_EXPOSURE", 0.50), -) -FEATURE_HARD_DEFENSE_EXPOSURE = FEATURE_RUNTIME_PARAMETERS.get( - "hard_defense_exposure", - getattr(STRATEGY_LOGIC, "DEFAULT_HARD_DEFENSE_EXPOSURE", 0.10), -) -FEATURE_SOFT_BREADTH_THRESHOLD = FEATURE_RUNTIME_PARAMETERS.get( - "soft_breadth_threshold", - getattr(STRATEGY_LOGIC, "DEFAULT_SOFT_BREADTH_THRESHOLD", 0.55), -) -FEATURE_HARD_BREADTH_THRESHOLD = FEATURE_RUNTIME_PARAMETERS.get( - "hard_breadth_threshold", - getattr(STRATEGY_LOGIC, "DEFAULT_HARD_BREADTH_THRESHOLD", 0.35), -) -FEATURE_RUNTIME_EXECUTION_WINDOW_TRADING_DAYS = FEATURE_RUNTIME_PARAMETERS.get( - "runtime_execution_window_trading_days", +STRATEGY_RUNTIME = load_strategy_runtime( + STRATEGY_PROFILE, + runtime_settings=RUNTIME_SETTINGS, + logger=lambda message: print(message, flush=True), ) -FEATURE_MIN_ADV20_USD = FEATURE_RUNTIME_PARAMETERS.get("min_adv20_usd") -FEATURE_SECTOR_WHITELIST = FEATURE_RUNTIME_PARAMETERS.get("sector_whitelist") -FEATURE_NORMALIZATION = FEATURE_RUNTIME_PARAMETERS.get("normalization") -FEATURE_SCORE_TEMPLATE = FEATURE_RUNTIME_PARAMETERS.get("score_template") -FEATURE_RESIDUAL_PROXY = FEATURE_RUNTIME_PARAMETERS.get("residual_proxy") -FEATURE_RUNTIME_CONFIG_NAME = FEATURE_RUNTIME_PARAMETERS.get( - "runtime_config_name", - RUNTIME_SETTINGS.strategy_profile, +STRATEGY_ENTRYPOINT = STRATEGY_RUNTIME.entrypoint +STRATEGY_SIGNAL_SOURCE = ( + "feature_snapshot" + if "feature_snapshot" in STRATEGY_RUNTIME.required_inputs + else "market_data" ) -FEATURE_RUNTIME_CONFIG_PATH = FEATURE_RUNTIME_PARAMETERS.get( - "runtime_config_path", - RUNTIME_SETTINGS.strategy_config_path, +STRATEGY_STATUS_ICON = STRATEGY_RUNTIME.status_icon +FEATURE_RUNTIME_PARAMETERS = dict(STRATEGY_RUNTIME.runtime_config) +STRATEGY_RUNTIME_CONFIG = dict(STRATEGY_RUNTIME.merged_runtime_config) +SAFE_HAVEN = str(STRATEGY_RUNTIME_CONFIG.get("safe_haven") or "BIL") +RANKING_POOL = list(STRATEGY_RUNTIME_CONFIG.get("ranking_pool", ())) +CANARY_ASSETS = list(STRATEGY_RUNTIME_CONFIG.get("canary_assets", ())) +TOP_N = STRATEGY_RUNTIME_CONFIG.get("top_n") +SMA_PERIOD = int(STRATEGY_RUNTIME_CONFIG.get("sma_period", 200)) +CANARY_BAD_THRESHOLD = STRATEGY_RUNTIME_CONFIG.get("canary_bad_threshold") +REBALANCE_MONTHS = STRATEGY_RUNTIME_CONFIG.get("rebalance_months") +FEATURE_SNAPSHOT_PATH = RUNTIME_SETTINGS.feature_snapshot_path +FEATURE_SNAPSHOT_MANIFEST_PATH = RUNTIME_SETTINGS.feature_snapshot_manifest_path +FEATURE_RUNTIME_CONFIG_PATH = ( + STRATEGY_RUNTIME_CONFIG.get("runtime_config_path") + or RUNTIME_SETTINGS.strategy_config_path ) -FEATURE_RUNTIME_CONFIG_SOURCE = FEATURE_RUNTIME_PARAMETERS.get( - "runtime_config_source", - RUNTIME_SETTINGS.strategy_config_source, +FEATURE_RUNTIME_CONFIG_SOURCE = ( + STRATEGY_RUNTIME_CONFIG.get("runtime_config_source") + or RUNTIME_SETTINGS.strategy_config_source ) RECONCILIATION_OUTPUT_PATH = RUNTIME_SETTINGS.reconciliation_output_path -strategy_check_sma = getattr(STRATEGY_LOGIC, "check_sma", None) -strategy_compute_13612w_momentum = getattr(STRATEGY_LOGIC, "compute_13612w_momentum", None) -strategy_compute_signals = STRATEGY_LOGIC.compute_signals TG_TOKEN = RUNTIME_SETTINGS.tg_token TG_CHAT_ID = RUNTIME_SETTINGS.tg_chat_id NOTIFY_LANG = RUNTIME_SETTINGS.notify_lang -DEFAULT_CASH_RESERVE_RATIO = 0.03 -CASH_RESERVE_RATIO = float( - FEATURE_RUNTIME_PARAMETERS.get( - "execution_cash_reserve_ratio", - DEFAULT_CASH_RESERVE_RATIO, - ) -) +CASH_RESERVE_RATIO = STRATEGY_RUNTIME.cash_reserve_ratio REBALANCE_THRESHOLD_RATIO = 0.02 # 2% of equity to trigger trades LIMIT_BUY_PREMIUM = 1.005 @@ -303,210 +224,19 @@ def get_historical_close(ib, symbol, duration="2 Y", bar_size="1 day"): # --------------------------------------------------------------------------- # Strategy logic # --------------------------------------------------------------------------- -def compute_13612w_momentum(closes, as_of_date=None): - if strategy_compute_13612w_momentum is None: - raise NotImplementedError(f"{STRATEGY_PROFILE} does not expose 13612W momentum") - return strategy_compute_13612w_momentum(closes, as_of_date=as_of_date) - - -def check_sma(closes, period=SMA_PERIOD): - if strategy_check_sma is None: - raise NotImplementedError(f"{STRATEGY_PROFILE} does not expose SMA filtering") - return strategy_check_sma(closes, period=period) - - def compute_signals(ib, current_holdings): - if STRATEGY_SIGNAL_SOURCE == "feature_snapshot": - run_as_of = resolve_run_as_of_date() - if not FEATURE_SNAPSHOT_PATH: - return ( - None, - "feature snapshot required", - False, - "fail_closed | reason=feature_snapshot_path_missing", - { - "strategy_profile": STRATEGY_PROFILE, - "feature_snapshot_path": None, - "strategy_config_path": FEATURE_RUNTIME_CONFIG_PATH, - "strategy_config_source": FEATURE_RUNTIME_CONFIG_SOURCE, - "dry_run_only": RUNTIME_SETTINGS.dry_run_only, - "snapshot_guard_decision": "fail_closed", - "fail_reason": "feature_snapshot_path_missing", - "managed_symbols": (), - "status_icon": "🛑", - }, - ) - guard_result = load_feature_snapshot_guarded( - FEATURE_SNAPSHOT_PATH, - run_as_of=run_as_of, - required_columns=getattr(STRATEGY_LOGIC, "REQUIRED_FEATURE_COLUMNS", ()), - snapshot_date_columns=getattr( - STRATEGY_LOGIC, - "SNAPSHOT_DATE_COLUMNS", - ("as_of", "snapshot_date"), - ), - max_snapshot_month_lag=int( - getattr(STRATEGY_LOGIC, "MAX_SNAPSHOT_MONTH_LAG", 1) - ), - manifest_path=FEATURE_SNAPSHOT_MANIFEST_PATH, - require_manifest=FEATURE_REQUIRE_SNAPSHOT_MANIFEST, - expected_strategy_profile=STRATEGY_PROFILE, - expected_config_name=FEATURE_RUNTIME_CONFIG_NAME, - expected_config_path=FEATURE_RUNTIME_CONFIG_PATH, - expected_contract_version=FEATURE_SNAPSHOT_CONTRACT_VERSION, - ) - guard_metadata = dict(guard_result.metadata) - print( - "snapshot_manifest_summary | " - f"profile={STRATEGY_PROFILE} decision={guard_metadata.get('snapshot_guard_decision')} " - f"snapshot_path={guard_metadata.get('snapshot_path')} " - f"snapshot_as_of={guard_metadata.get('snapshot_as_of')} " - f"snapshot_age_days={guard_metadata.get('snapshot_age_days')} " - f"snapshot_file_ts={guard_metadata.get('snapshot_file_timestamp')} " - f"manifest_path={guard_metadata.get('snapshot_manifest_path')} " - f"manifest_exists={guard_metadata.get('snapshot_manifest_exists')} " - f"manifest_contract={guard_metadata.get('snapshot_manifest_contract_version')} " - f"expected_config={FEATURE_RUNTIME_CONFIG_PATH} " - f"expected_profile={STRATEGY_PROFILE}", - flush=True, - ) - if guard_result.metadata.get("snapshot_guard_decision") != "proceed": - decision = guard_metadata.get("snapshot_guard_decision") - reason = guard_metadata.get("fail_reason") or guard_metadata.get("no_op_reason") - return ( - None, - "feature snapshot guard blocked execution", - False, - f"{decision} | reason={reason}", - { - "strategy_profile": STRATEGY_PROFILE, - "strategy_config_path": FEATURE_RUNTIME_CONFIG_PATH, - "strategy_config_source": FEATURE_RUNTIME_CONFIG_SOURCE, - "dry_run_only": RUNTIME_SETTINGS.dry_run_only, - "managed_symbols": (), - "status_icon": "🛑", - **guard_metadata, - }, - ) - feature_snapshot = guard_result.frame - feature_kwargs = { - "benchmark_symbol": FEATURE_BENCHMARK_SYMBOL, - "safe_haven": SAFE_HAVEN, - "holdings_count": FEATURE_HOLDINGS_COUNT, - "single_name_cap": FEATURE_SINGLE_NAME_CAP, - "sector_cap": FEATURE_SECTOR_CAP, - "hold_bonus": HOLD_BONUS, - "risk_on_exposure": FEATURE_RISK_ON_EXPOSURE, - "soft_defense_exposure": FEATURE_SOFT_DEFENSE_EXPOSURE, - "hard_defense_exposure": FEATURE_HARD_DEFENSE_EXPOSURE, - "soft_breadth_threshold": FEATURE_SOFT_BREADTH_THRESHOLD, - "hard_breadth_threshold": FEATURE_HARD_BREADTH_THRESHOLD, - "min_adv20_usd": FEATURE_MIN_ADV20_USD, - "sector_whitelist": FEATURE_SECTOR_WHITELIST, - "normalization": FEATURE_NORMALIZATION, - "score_template": FEATURE_SCORE_TEMPLATE, - "run_as_of": run_as_of, - "runtime_execution_window_trading_days": FEATURE_RUNTIME_EXECUTION_WINDOW_TRADING_DAYS, - "runtime_config_name": FEATURE_RUNTIME_CONFIG_NAME, - "runtime_config_path": FEATURE_RUNTIME_CONFIG_PATH, - "runtime_config_source": FEATURE_RUNTIME_CONFIG_SOURCE, - "residual_proxy": FEATURE_RESIDUAL_PROXY, - } - feature_kwargs = { - key: value - for key, value in feature_kwargs.items() - if key in FEATURE_SIGNAL_KWARG_KEYS and value is not None - } - try: - result = strategy_compute_signals( - feature_snapshot, - current_holdings, - **feature_kwargs, - ) - except Exception as exc: - return ( - None, - "feature snapshot compute failed", - False, - f"fail_closed | reason=feature_snapshot_compute_failed:{type(exc).__name__}:{exc}", - { - "strategy_profile": STRATEGY_PROFILE, - "strategy_config_path": FEATURE_RUNTIME_CONFIG_PATH, - "strategy_config_source": FEATURE_RUNTIME_CONFIG_SOURCE, - "dry_run_only": RUNTIME_SETTINGS.dry_run_only, - "managed_symbols": (), - "status_icon": "🛑", - **guard_metadata, - "snapshot_guard_decision": "fail_closed", - "fail_reason": f"feature_snapshot_compute_failed:{type(exc).__name__}:{exc}", - }, - ) - if len(result) == 5: - target_weights, signal_desc, is_emergency, status_desc, metadata = result - return ( - target_weights, - signal_desc, - is_emergency, - status_desc, - { - "strategy_profile": STRATEGY_PROFILE, - "feature_snapshot_path": FEATURE_SNAPSHOT_PATH, - "strategy_config_path": FEATURE_RUNTIME_CONFIG_PATH, - "strategy_config_source": FEATURE_RUNTIME_CONFIG_SOURCE, - "safe_haven_symbol": SAFE_HAVEN, - "dry_run_only": RUNTIME_SETTINGS.dry_run_only, - "trade_date": run_as_of.date().isoformat(), - **guard_metadata, - **metadata, - }, - ) - target_weights, signal_desc, is_emergency, status_desc = result - return ( - target_weights, - signal_desc, - is_emergency, - status_desc, - { - "strategy_profile": STRATEGY_PROFILE, - "feature_snapshot_path": FEATURE_SNAPSHOT_PATH, - "strategy_config_path": FEATURE_RUNTIME_CONFIG_PATH, - "strategy_config_source": FEATURE_RUNTIME_CONFIG_SOURCE, - "safe_haven_symbol": SAFE_HAVEN, - "dry_run_only": RUNTIME_SETTINGS.dry_run_only, - "trade_date": run_as_of.date().isoformat(), - **guard_metadata, - "managed_symbols": tuple( - getattr( - STRATEGY_LOGIC, - "extract_managed_symbols", - )(feature_snapshot, benchmark_symbol=FEATURE_BENCHMARK_SYMBOL, safe_haven=SAFE_HAVEN) - ), - "status_icon": STRATEGY_STATUS_ICON, - }, - ) - - return strategy_compute_signals( - ib, - current_holdings, - get_historical_close=get_historical_close, - ranking_pool=RANKING_POOL, - canary_assets=CANARY_ASSETS, - safe_haven=SAFE_HAVEN, - top_n=TOP_N, - hold_bonus=HOLD_BONUS, - canary_bad_threshold=CANARY_BAD_THRESHOLD, - rebalance_months=REBALANCE_MONTHS, + evaluation = STRATEGY_RUNTIME.evaluate( + ib=ib, + current_holdings=current_holdings, + historical_close_loader=get_historical_close, + run_as_of=resolve_run_as_of_date(), translator=t, pacing_sec=HIST_DATA_PACING_SEC, - sma_period=SMA_PERIOD, - ) + ( - { - "strategy_profile": STRATEGY_PROFILE, - "managed_symbols": tuple(RANKING_POOL + [SAFE_HAVEN]), - "status_icon": STRATEGY_STATUS_ICON, - "safe_haven_symbol": SAFE_HAVEN, - "dry_run_only": RUNTIME_SETTINGS.dry_run_only, - }, + ) + return map_strategy_decision( + evaluation.decision, + strategy_profile=STRATEGY_PROFILE, + runtime_metadata=evaluation.metadata, ) diff --git a/requirements.txt b/requirements.txt index 31c9f44..8c5b981 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,7 +1,7 @@ flask gunicorn -quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@6e8cc058b821aea8a54015d4b39e02fbdd3dc198 -us-equity-strategies @ git+https://github.com/QuantStrategyLab/UsEquityStrategies.git@53f996ef489700f325a5b750355d644f4aed7270 +quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@5174d9e40f79fffae47450a42e26434145d28b31 +us-equity-strategies @ git+https://github.com/QuantStrategyLab/UsEquityStrategies.git@84da76d1ed17e9cf8bec33d9f1f8020f61362a78 pandas numpy requests diff --git a/strategy_loader.py b/strategy_loader.py index 762f0c2..18d755c 100644 --- a/strategy_loader.py +++ b/strategy_loader.py @@ -1,18 +1,35 @@ from __future__ import annotations -from types import ModuleType - -from quant_platform_kit.common.strategies import load_strategy_component_module +from quant_platform_kit.common.strategies import ( + StrategyDefinition, + load_strategy_entrypoint, +) +from quant_platform_kit.strategy_contracts import StrategyEntrypoint, StrategyRuntimeAdapter +from us_equity_strategies import get_platform_runtime_adapter from strategy_registry import IBKR_PLATFORM, resolve_strategy_definition -def load_signal_logic_module(raw_profile: str | None) -> ModuleType: - definition = resolve_strategy_definition( +def load_strategy_definition(raw_profile: str | None) -> StrategyDefinition: + return resolve_strategy_definition( raw_profile, platform_id=IBKR_PLATFORM, ) - return load_strategy_component_module( + + +def load_strategy_entrypoint_for_profile(raw_profile: str | None) -> StrategyEntrypoint: + definition = load_strategy_definition(raw_profile) + return load_strategy_entrypoint( definition, - component_name="signal_logic", + platform_id=IBKR_PLATFORM, + available_inputs=("historical_close_loader", "feature_snapshot"), + available_capabilities=("broker_client",), + ) + + +def load_strategy_runtime_adapter_for_profile(raw_profile: str | None) -> StrategyRuntimeAdapter: + definition = load_strategy_definition(raw_profile) + return get_platform_runtime_adapter( + definition.profile, + platform_id=IBKR_PLATFORM, ) diff --git a/strategy_runtime.py b/strategy_runtime.py new file mode 100644 index 0000000..ed460d3 --- /dev/null +++ b/strategy_runtime.py @@ -0,0 +1,351 @@ +from __future__ import annotations + +from collections.abc import Callable, Mapping +from dataclasses import dataclass, field +from typing import Any + +import pandas as pd + +from application.feature_snapshot_service import load_feature_snapshot_guarded +from quant_platform_kit.strategy_contracts import ( + StrategyContext, + StrategyDecision, + StrategyEntrypoint, + StrategyRuntimeAdapter, +) +from runtime_config_support import PlatformRuntimeSettings +from strategy_loader import ( + load_strategy_definition, + load_strategy_entrypoint_for_profile, + load_strategy_runtime_adapter_for_profile, +) + + +DEFAULT_CASH_RESERVE_RATIO = 0.03 +_FEATURE_SNAPSHOT_INPUT = "feature_snapshot" +_HISTORICAL_CLOSE_INPUT = "historical_close_loader" + + +@dataclass(frozen=True) +class StrategyEvaluationResult: + decision: StrategyDecision + metadata: Mapping[str, Any] = field(default_factory=dict) + + +@dataclass(frozen=True) +class LoadedStrategyRuntime: + entrypoint: StrategyEntrypoint + runtime_settings: PlatformRuntimeSettings + runtime_adapter: StrategyRuntimeAdapter + runtime_config: Mapping[str, Any] = field(default_factory=dict) + merged_runtime_config: Mapping[str, Any] = field(default_factory=dict) + status_icon: str = "🐤" + cash_reserve_ratio: float = DEFAULT_CASH_RESERVE_RATIO + logger: Callable[[str], None] = print + + @property + def profile(self) -> str: + return self.entrypoint.manifest.profile + + @property + def required_inputs(self) -> frozenset[str]: + return frozenset(self.entrypoint.manifest.required_inputs) + + def evaluate( + self, + *, + ib, + current_holdings, + historical_close_loader: Callable[..., Any], + run_as_of: pd.Timestamp, + translator: Callable[[str], str], + pacing_sec: float, + ) -> StrategyEvaluationResult: + run_as_of = pd.Timestamp(run_as_of).normalize() + if _FEATURE_SNAPSHOT_INPUT in self.required_inputs: + return self._evaluate_feature_snapshot_strategy( + current_holdings=current_holdings, + run_as_of=run_as_of, + ) + if _HISTORICAL_CLOSE_INPUT in self.required_inputs: + return self._evaluate_market_data_strategy( + ib=ib, + current_holdings=current_holdings, + historical_close_loader=historical_close_loader, + run_as_of=run_as_of, + translator=translator, + pacing_sec=pacing_sec, + ) + raise ValueError( + f"Unsupported required_inputs for IBKR strategy profile {self.profile!r}: " + f"{', '.join(sorted(self.required_inputs)) or ''}" + ) + + def _evaluate_market_data_strategy( + self, + *, + ib, + current_holdings, + historical_close_loader: Callable[..., Any], + run_as_of: pd.Timestamp, + translator: Callable[[str], str], + pacing_sec: float, + ) -> StrategyEvaluationResult: + runtime_config = dict(self.runtime_config) + runtime_config.setdefault("translator", translator) + runtime_config.setdefault("pacing_sec", float(pacing_sec)) + ctx = StrategyContext( + as_of=run_as_of, + market_data={"historical_close_loader": historical_close_loader}, + state={"current_holdings": tuple(current_holdings)}, + runtime_config=runtime_config, + capabilities={"broker_client": ib}, + ) + decision = self.entrypoint.evaluate(ctx) + safe_haven_symbol = str(self.merged_runtime_config.get("safe_haven") or "").strip().upper() or None + ranking_pool = tuple(str(symbol) for symbol in self.merged_runtime_config.get("ranking_pool", ())) + managed_candidates = list(ranking_pool) + if safe_haven_symbol: + managed_candidates.append(safe_haven_symbol) + managed_symbols = tuple(dict.fromkeys(managed_candidates)) + metadata = { + "strategy_profile": self.profile, + "managed_symbols": managed_symbols, + "status_icon": self.status_icon, + "dry_run_only": self.runtime_settings.dry_run_only, + } + if safe_haven_symbol: + metadata["safe_haven_symbol"] = safe_haven_symbol + return StrategyEvaluationResult(decision=decision, metadata=metadata) + + def _evaluate_feature_snapshot_strategy( + self, + *, + current_holdings, + run_as_of: pd.Timestamp, + ) -> StrategyEvaluationResult: + if not self.runtime_settings.feature_snapshot_path: + metadata = { + "strategy_profile": self.profile, + "feature_snapshot_path": None, + "strategy_config_path": self.runtime_settings.strategy_config_path, + "strategy_config_source": self.runtime_settings.strategy_config_source, + "dry_run_only": self.runtime_settings.dry_run_only, + "snapshot_guard_decision": "fail_closed", + "fail_reason": "feature_snapshot_path_missing", + "managed_symbols": (), + "status_icon": "🛑", + } + decision = StrategyDecision( + risk_flags=("no_execute",), + diagnostics={ + "signal_description": "feature snapshot required", + "status_description": "fail_closed | reason=feature_snapshot_path_missing", + "actionable": False, + "snapshot_guard_decision": "fail_closed", + "fail_reason": "feature_snapshot_path_missing", + }, + ) + return StrategyEvaluationResult(decision=decision, metadata=metadata) + + runtime_config_name = str( + self.merged_runtime_config.get("runtime_config_name") + or self.runtime_settings.strategy_profile + ) + runtime_config_path = self.merged_runtime_config.get("runtime_config_path") or self.runtime_settings.strategy_config_path + runtime_config_source = self.merged_runtime_config.get("runtime_config_source") or self.runtime_settings.strategy_config_source + benchmark_symbol = str(self.merged_runtime_config.get("benchmark_symbol") or "SPY").strip().upper() + safe_haven_symbol = str(self.merged_runtime_config.get("safe_haven") or "BOXX").strip().upper() + + guard_result = load_feature_snapshot_guarded( + self.runtime_settings.feature_snapshot_path, + run_as_of=run_as_of, + required_columns=self._required_feature_columns(), + snapshot_date_columns=self._snapshot_date_columns(), + max_snapshot_month_lag=self._max_snapshot_month_lag(), + manifest_path=self.runtime_settings.feature_snapshot_manifest_path, + require_manifest=self._require_snapshot_manifest(), + expected_strategy_profile=self.profile, + expected_config_name=runtime_config_name, + expected_config_path=runtime_config_path, + expected_contract_version=self._snapshot_contract_version(), + ) + guard_metadata = dict(guard_result.metadata) + self.logger( + "snapshot_manifest_summary | " + f"profile={self.profile} decision={guard_metadata.get('snapshot_guard_decision')} " + f"snapshot_path={guard_metadata.get('snapshot_path')} " + f"snapshot_as_of={guard_metadata.get('snapshot_as_of')} " + f"snapshot_age_days={guard_metadata.get('snapshot_age_days')} " + f"snapshot_file_ts={guard_metadata.get('snapshot_file_timestamp')} " + f"manifest_path={guard_metadata.get('snapshot_manifest_path')} " + f"manifest_exists={guard_metadata.get('snapshot_manifest_exists')} " + f"manifest_contract={guard_metadata.get('snapshot_manifest_contract_version')} " + f"expected_config={runtime_config_path} " + f"expected_profile={self.profile}" + ) + if guard_metadata.get("snapshot_guard_decision") != "proceed": + decision_text = str(guard_metadata.get("snapshot_guard_decision") or "fail_closed") + reason = guard_metadata.get("fail_reason") or guard_metadata.get("no_op_reason") + metadata = { + "strategy_profile": self.profile, + "strategy_config_path": runtime_config_path, + "strategy_config_source": runtime_config_source, + "dry_run_only": self.runtime_settings.dry_run_only, + "managed_symbols": (), + "status_icon": "🛑", + **guard_metadata, + } + decision = StrategyDecision( + risk_flags=("no_execute",), + diagnostics={ + "signal_description": "feature snapshot guard blocked execution", + "status_description": f"{decision_text} | reason={reason}", + "actionable": False, + "snapshot_guard_decision": decision_text, + "fail_reason": guard_metadata.get("fail_reason"), + "no_op_reason": guard_metadata.get("no_op_reason"), + }, + ) + return StrategyEvaluationResult(decision=decision, metadata=metadata) + + feature_snapshot = guard_result.frame + managed_symbols = self._extract_managed_symbols( + feature_snapshot, + benchmark_symbol=benchmark_symbol, + safe_haven_symbol=safe_haven_symbol, + ) + ctx = StrategyContext( + as_of=run_as_of, + market_data={"feature_snapshot": feature_snapshot}, + state={"current_holdings": tuple(current_holdings)}, + runtime_config=dict(self.runtime_config), + ) + try: + decision = self.entrypoint.evaluate(ctx) + except Exception as exc: + fail_reason = f"feature_snapshot_compute_failed:{type(exc).__name__}:{exc}" + metadata = { + "strategy_profile": self.profile, + "strategy_config_path": runtime_config_path, + "strategy_config_source": runtime_config_source, + "dry_run_only": self.runtime_settings.dry_run_only, + "managed_symbols": (), + "status_icon": "🛑", + **guard_metadata, + "snapshot_guard_decision": "fail_closed", + "fail_reason": fail_reason, + } + decision = StrategyDecision( + risk_flags=("no_execute",), + diagnostics={ + "signal_description": "feature snapshot compute failed", + "status_description": f"fail_closed | reason={fail_reason}", + "actionable": False, + "snapshot_guard_decision": "fail_closed", + "fail_reason": fail_reason, + }, + ) + return StrategyEvaluationResult(decision=decision, metadata=metadata) + metadata = { + "strategy_profile": self.profile, + "feature_snapshot_path": self.runtime_settings.feature_snapshot_path, + "strategy_config_path": runtime_config_path, + "strategy_config_source": runtime_config_source, + "safe_haven_symbol": safe_haven_symbol, + "dry_run_only": self.runtime_settings.dry_run_only, + "trade_date": run_as_of.date().isoformat(), + "managed_symbols": managed_symbols, + "status_icon": self.status_icon, + **guard_metadata, + } + return StrategyEvaluationResult(decision=decision, metadata=metadata) + + def _extract_managed_symbols( + self, + feature_snapshot, + *, + benchmark_symbol: str, + safe_haven_symbol: str, + ) -> tuple[str, ...]: + extractor = self.runtime_adapter.managed_symbols_extractor + if extractor is None: + if safe_haven_symbol: + return (safe_haven_symbol,) + return () + if callable(extractor): + return tuple( + extractor( + feature_snapshot, + benchmark_symbol=benchmark_symbol, + safe_haven=safe_haven_symbol, + ) + ) + if safe_haven_symbol: + return (safe_haven_symbol,) + return () + + def _required_feature_columns(self) -> tuple[str, ...] | frozenset[str]: + return self.runtime_adapter.required_feature_columns + + def _snapshot_date_columns(self) -> tuple[str, ...]: + return tuple(self.runtime_adapter.snapshot_date_columns) + + def _max_snapshot_month_lag(self) -> int: + return int(self.runtime_adapter.max_snapshot_month_lag) + + def _require_snapshot_manifest(self) -> bool: + return bool(self.runtime_adapter.require_snapshot_manifest) + + def _snapshot_contract_version(self) -> str | None: + return self.runtime_adapter.snapshot_contract_version + + def load_runtime_parameters(self) -> dict[str, Any]: + runtime_loader = self.runtime_adapter.runtime_parameter_loader + if not callable(runtime_loader): + return {} + return dict( + runtime_loader( + config_path=self.runtime_settings.strategy_config_path, + logger=self.logger, + ) + or {} + ) + + +def load_strategy_runtime( + raw_profile: str | None, + *, + runtime_settings: PlatformRuntimeSettings, + logger: Callable[[str], None], +) -> LoadedStrategyRuntime: + strategy_definition = load_strategy_definition(raw_profile) + entrypoint = load_strategy_entrypoint_for_profile(strategy_definition.profile) + runtime_adapter = load_strategy_runtime_adapter_for_profile(strategy_definition.profile) + runtime = LoadedStrategyRuntime( + entrypoint=entrypoint, + runtime_adapter=runtime_adapter, + runtime_settings=runtime_settings, + logger=logger, + ) + runtime_config: dict[str, Any] = {} + if _FEATURE_SNAPSHOT_INPUT in frozenset(entrypoint.manifest.required_inputs): + runtime_config = runtime.load_runtime_parameters() + + merged_runtime_config = dict(entrypoint.manifest.default_config) + merged_runtime_config.update(runtime_config) + return LoadedStrategyRuntime( + entrypoint=entrypoint, + runtime_adapter=runtime_adapter, + runtime_settings=runtime_settings, + runtime_config=runtime_config, + merged_runtime_config=merged_runtime_config, + status_icon=runtime_adapter.status_icon, + cash_reserve_ratio=float( + merged_runtime_config.get( + "execution_cash_reserve_ratio", + DEFAULT_CASH_RESERVE_RATIO, + ) + ), + logger=logger, + ) diff --git a/tests/test_decision_mapper.py b/tests/test_decision_mapper.py new file mode 100644 index 0000000..1c1b259 --- /dev/null +++ b/tests/test_decision_mapper.py @@ -0,0 +1,57 @@ +from quant_platform_kit.strategy_contracts import PositionTarget, StrategyDecision + +from decision_mapper import map_strategy_decision + + +def test_map_strategy_decision_maps_weight_positions_and_safe_haven(): + decision = StrategyDecision( + positions=( + PositionTarget(symbol="AAA", target_weight=0.6), + PositionTarget(symbol="BOXX", target_weight=0.4, role="safe_haven"), + ), + diagnostics={ + "signal_description": "risk on", + "status_description": "breadth=60.0%", + }, + ) + + target_weights, signal_desc, is_emergency, status_desc, metadata = map_strategy_decision( + decision, + strategy_profile="tech_pullback_cash_buffer", + runtime_metadata={"status_icon": "🧲", "dry_run_only": True}, + ) + + assert target_weights == {"AAA": 0.6, "BOXX": 0.4} + assert signal_desc == "risk on" + assert is_emergency is False + assert status_desc == "breadth=60.0%" + assert metadata["safe_haven_symbol"] == "BOXX" + assert metadata["managed_symbols"] == ("AAA", "BOXX") + assert metadata["status_icon"] == "🧲" + + +def test_map_strategy_decision_returns_noop_when_flagged_no_execute(): + decision = StrategyDecision( + risk_flags=("no_execute",), + diagnostics={ + "signal_description": "feature snapshot guard blocked execution", + "status_description": "fail_closed | reason=feature_snapshot_path_missing", + }, + ) + + target_weights, signal_desc, is_emergency, status_desc, metadata = map_strategy_decision( + decision, + strategy_profile="tech_pullback_cash_buffer", + runtime_metadata={ + "status_icon": "🛑", + "snapshot_guard_decision": "fail_closed", + "managed_symbols": (), + }, + ) + + assert target_weights is None + assert signal_desc == "feature snapshot guard blocked execution" + assert is_emergency is False + assert status_desc == "fail_closed | reason=feature_snapshot_path_missing" + assert metadata["actionable"] is False + assert metadata["snapshot_guard_decision"] == "fail_closed" diff --git a/tests/test_snapshot_strategy_runtime.py b/tests/test_snapshot_strategy_runtime.py index 9f5f692..d590f1b 100644 --- a/tests/test_snapshot_strategy_runtime.py +++ b/tests/test_snapshot_strategy_runtime.py @@ -4,6 +4,8 @@ import json from types import SimpleNamespace +import strategy_runtime as strategy_runtime_module + def _sha256_file(path: Path) -> str: hasher = hashlib.sha256() @@ -71,17 +73,10 @@ def fake_load_feature_snapshot_guarded(path, **_kwargs): }, ) - monkeypatch.setattr(module, "load_feature_snapshot_guarded", fake_load_feature_snapshot_guarded) monkeypatch.setattr( - module, - "strategy_compute_signals", - lambda snapshot, holdings, **kwargs: ( - {"BOXX": 1.0}, - "signal", - False, - "breadth=0.0%", - {"managed_symbols": ("BOXX",), "status_icon": "📏"}, - ), + strategy_runtime_module, + "load_feature_snapshot_guarded", + fake_load_feature_snapshot_guarded, ) result = module.compute_signals(None, {"AAA"}) diff --git a/tests/test_strategy_loader.py b/tests/test_strategy_loader.py index d907803..45e0916 100644 --- a/tests/test_strategy_loader.py +++ b/tests/test_strategy_loader.py @@ -3,34 +3,40 @@ import pytest -from strategy_loader import load_signal_logic_module +from strategy_loader import ( + load_strategy_entrypoint_for_profile, + load_strategy_runtime_adapter_for_profile, +) -def test_load_signal_logic_module_resolves_global_etf_rotation(monkeypatch): +def test_load_strategy_entrypoint_for_profile_resolves_global_etf_rotation(monkeypatch): market_calendars_module = types.ModuleType("pandas_market_calendars") market_calendars_module.get_calendar = lambda name: None monkeypatch.setitem(sys.modules, "pandas_market_calendars", market_calendars_module) - sys.modules.pop("us_equity_strategies.strategies.global_etf_rotation", None) - module = load_signal_logic_module("global_etf_rotation") + entrypoint = load_strategy_entrypoint_for_profile("global_etf_rotation") - assert module.__name__ == "us_equity_strategies.strategies.global_etf_rotation" - assert module.TOP_N == 2 + assert entrypoint.manifest.profile == "global_etf_rotation" + assert "historical_close_loader" in entrypoint.manifest.required_inputs -def test_load_signal_logic_module_resolves_russell_1000_multi_factor_defensive(): +def test_load_strategy_entrypoint_for_profile_resolves_tech_pullback_cash_buffer(monkeypatch): try: import pandas # noqa: F401 except ModuleNotFoundError: return - module = load_signal_logic_module("russell_1000_multi_factor_defensive") + market_calendars_module = types.ModuleType("pandas_market_calendars") + market_calendars_module.get_calendar = lambda name: None + monkeypatch.setitem(sys.modules, "pandas_market_calendars", market_calendars_module) + + entrypoint = load_strategy_entrypoint_for_profile("tech_pullback_cash_buffer") - assert module.__name__ == "us_equity_strategies.strategies.russell_1000_multi_factor_defensive" - assert module.SIGNAL_SOURCE == "feature_snapshot" + assert entrypoint.manifest.profile == "tech_pullback_cash_buffer" + assert entrypoint.manifest.default_config["safe_haven"] == "BOXX" -def test_load_signal_logic_module_resolves_tech_pullback_cash_buffer(monkeypatch): +def test_load_strategy_entrypoint_for_profile_rejects_legacy_cash_buffer_profile(monkeypatch): try: import pandas # noqa: F401 except ModuleNotFoundError: @@ -39,15 +45,12 @@ def test_load_signal_logic_module_resolves_tech_pullback_cash_buffer(monkeypatch market_calendars_module = types.ModuleType("pandas_market_calendars") market_calendars_module.get_calendar = lambda name: None monkeypatch.setitem(sys.modules, "pandas_market_calendars", market_calendars_module) - sys.modules.pop("us_equity_strategies.strategies.tech_pullback_cash_buffer", None) - module = load_signal_logic_module("tech_pullback_cash_buffer") - - assert module.__name__ == "us_equity_strategies.strategies.tech_pullback_cash_buffer" - assert module.SIGNAL_SOURCE == "feature_snapshot" + with pytest.raises(ValueError, match="Unsupported STRATEGY_PROFILE"): + load_strategy_entrypoint_for_profile("cash_buffer_branch_default") -def test_load_signal_logic_module_rejects_legacy_cash_buffer_profile(monkeypatch): +def test_load_strategy_runtime_adapter_for_profile_resolves_tech_pullback_cash_buffer(monkeypatch): try: import pandas # noqa: F401 except ModuleNotFoundError: @@ -56,7 +59,9 @@ def test_load_signal_logic_module_rejects_legacy_cash_buffer_profile(monkeypatch market_calendars_module = types.ModuleType("pandas_market_calendars") market_calendars_module.get_calendar = lambda name: None monkeypatch.setitem(sys.modules, "pandas_market_calendars", market_calendars_module) - sys.modules.pop("us_equity_strategies.strategies.tech_pullback_cash_buffer", None) - with pytest.raises(ValueError, match="Unsupported STRATEGY_PROFILE"): - load_signal_logic_module("cash_buffer_branch_default") + adapter = load_strategy_runtime_adapter_for_profile("tech_pullback_cash_buffer") + + assert adapter.status_icon == "🧲" + assert adapter.require_snapshot_manifest is True + assert adapter.snapshot_contract_version == "tech_pullback_cash_buffer.feature_snapshot.v1" diff --git a/tests/test_strategy_runtime.py b/tests/test_strategy_runtime.py new file mode 100644 index 0000000..e435dd4 --- /dev/null +++ b/tests/test_strategy_runtime.py @@ -0,0 +1,270 @@ +from types import SimpleNamespace + +import strategy_runtime as strategy_runtime_module +from quant_platform_kit.strategy_contracts import ( + PositionTarget, + StrategyDecision, + StrategyManifest, + StrategyRuntimeAdapter, +) +from runtime_config_support import PlatformRuntimeSettings + + +def _build_runtime_settings(profile: str = "tech_pullback_cash_buffer") -> PlatformRuntimeSettings: + return PlatformRuntimeSettings( + project_id=None, + ib_gateway_instance_name="127.0.0.1", + ib_gateway_zone="", + ib_gateway_mode="live", + ib_gateway_ip_mode="internal", + ib_client_id=1, + strategy_profile=profile, + strategy_domain="us_equity", + feature_snapshot_path="/tmp/snapshot.csv", + feature_snapshot_manifest_path=None, + strategy_config_path="/tmp/config.json", + strategy_config_source="env", + reconciliation_output_path=None, + dry_run_only=True, + account_group="default", + service_name=None, + account_ids=(), + tg_token=None, + tg_chat_id=None, + notify_lang="en", + ) + + +def test_main_compute_signals_uses_strategy_runtime_decision(strategy_module, monkeypatch): + observed = {} + + class FakeRuntime: + def evaluate( + self, + *, + ib, + current_holdings, + historical_close_loader, + run_as_of, + translator, + pacing_sec, + ): + observed["ib"] = ib + observed["current_holdings"] = tuple(sorted(current_holdings)) + observed["run_as_of"] = str(run_as_of.date()) + observed["pacing_sec"] = pacing_sec + observed["translator_sample"] = translator("equity") + observed["historical_loader"] = historical_close_loader + return type( + "Evaluation", + (), + { + "decision": StrategyDecision( + positions=( + PositionTarget(symbol="AAA", target_weight=0.8), + PositionTarget(symbol="BIL", target_weight=0.2, role="safe_haven"), + ), + diagnostics={ + "signal_description": "rotation signal", + "status_description": "canary=ok", + }, + ), + "metadata": { + "strategy_profile": "global_etf_rotation", + "managed_symbols": ("AAA", "BIL"), + "status_icon": "🐤", + "dry_run_only": False, + }, + }, + )() + + monkeypatch.setattr(strategy_module, "STRATEGY_RUNTIME", FakeRuntime()) + monkeypatch.setattr(strategy_module, "resolve_run_as_of_date", lambda: strategy_module.pd.Timestamp("2026-04-07")) + + result = strategy_module.compute_signals("fake-ib", {"AAA"}) + + assert result[0] == {"AAA": 0.8, "BIL": 0.2} + assert result[1] == "rotation signal" + assert result[2] is False + assert result[3] == "canary=ok" + assert result[4]["managed_symbols"] == ("AAA", "BIL") + assert observed["ib"] == "fake-ib" + assert observed["current_holdings"] == ("AAA",) + assert observed["run_as_of"] == "2026-04-07" + assert observed["pacing_sec"] == strategy_module.HIST_DATA_PACING_SEC + assert observed["translator_sample"] + + +def test_load_strategy_runtime_uses_entrypoint_defaults_and_runtime_adapter(monkeypatch): + class FakeEntrypoint: + manifest = StrategyManifest( + profile="tech_pullback_cash_buffer", + domain="us_equity", + display_name="Tech Pullback Cash Buffer", + description="test", + required_inputs=frozenset({"feature_snapshot"}), + default_config={"safe_haven": "BOXX", "benchmark_symbol": "QQQ"}, + ) + + def evaluate(self, ctx): + return StrategyDecision() + + monkeypatch.setattr( + strategy_runtime_module, + "load_strategy_definition", + lambda raw_profile: SimpleNamespace(profile="tech_pullback_cash_buffer"), + ) + monkeypatch.setattr( + strategy_runtime_module, + "load_strategy_entrypoint_for_profile", + lambda raw_profile: FakeEntrypoint(), + ) + monkeypatch.setattr( + strategy_runtime_module, + "load_strategy_runtime_adapter_for_profile", + lambda raw_profile: StrategyRuntimeAdapter( + status_icon="🧲", + runtime_parameter_loader=lambda **_kwargs: { + "benchmark_symbol": "SPY", + "rebalance_months": (1, 4, 7, 10), + }, + ), + ) + + runtime = strategy_runtime_module.load_strategy_runtime( + "tech_pullback_cash_buffer", + runtime_settings=_build_runtime_settings(), + logger=lambda _message: None, + ) + + assert runtime.entrypoint.manifest.profile == "tech_pullback_cash_buffer" + assert runtime.runtime_config["benchmark_symbol"] == "SPY" + assert runtime.merged_runtime_config["safe_haven"] == "BOXX" + assert runtime.merged_runtime_config["benchmark_symbol"] == "SPY" + assert runtime.merged_runtime_config["rebalance_months"] == (1, 4, 7, 10) + assert runtime.status_icon == "🧲" + + +def test_feature_snapshot_runtime_prefers_unified_runtime_adapter_metadata(monkeypatch): + captured = {} + + class FakeEntrypoint: + manifest = StrategyManifest( + profile="tech_pullback_cash_buffer", + domain="us_equity", + display_name="Tech Pullback Cash Buffer", + description="test", + required_inputs=frozenset({"feature_snapshot"}), + default_config={"safe_haven": "BOXX", "benchmark_symbol": "QQQ"}, + ) + + def evaluate(self, ctx): + return StrategyDecision() + + runtime = strategy_runtime_module.LoadedStrategyRuntime( + entrypoint=FakeEntrypoint(), + runtime_adapter=StrategyRuntimeAdapter( + status_icon="🧲", + required_feature_columns=frozenset({"symbol", "close"}), + snapshot_date_columns=("as_of",), + max_snapshot_month_lag=2, + require_snapshot_manifest=True, + snapshot_contract_version="adapter.contract", + managed_symbols_extractor=lambda *_args, **_kwargs: ("AAPL", "BOXX"), + ), + runtime_settings=_build_runtime_settings(), + runtime_config={}, + merged_runtime_config={"safe_haven": "BOXX", "benchmark_symbol": "QQQ"}, + status_icon="🧲", + logger=lambda _message: None, + ) + + def fake_guard(path, **kwargs): + captured.update(kwargs) + return SimpleNamespace( + frame=[{"as_of": "2026-03-31", "symbol": "AAPL", "close": 1.0}], + metadata={ + "snapshot_guard_decision": "proceed", + "snapshot_as_of": "2026-03-31", + "snapshot_path": path, + "snapshot_age_days": 1, + }, + ) + + monkeypatch.setattr(strategy_runtime_module, "load_feature_snapshot_guarded", fake_guard) + + result = runtime.evaluate( + ib=None, + current_holdings={"AAPL"}, + historical_close_loader=lambda *_args, **_kwargs: None, + run_as_of=strategy_runtime_module.pd.Timestamp("2026-04-01"), + translator=lambda key, **_kwargs: key, + pacing_sec=0.5, + ) + + assert captured["required_columns"] == frozenset({"symbol", "close"}) + assert captured["snapshot_date_columns"] == ("as_of",) + assert captured["max_snapshot_month_lag"] == 2 + assert captured["require_manifest"] is True + assert captured["expected_contract_version"] == "adapter.contract" + assert result.metadata["managed_symbols"] == ("AAPL", "BOXX") + + +def test_feature_snapshot_runtime_fail_closes_on_entrypoint_exception(monkeypatch): + class ExplodingEntrypoint: + manifest = StrategyManifest( + profile="tech_pullback_cash_buffer", + domain="us_equity", + display_name="Tech Pullback Cash Buffer", + description="test", + required_inputs=frozenset({"feature_snapshot"}), + default_config={"safe_haven": "BOXX"}, + ) + + def evaluate(self, ctx): + raise RuntimeError("boom") + + runtime = strategy_runtime_module.LoadedStrategyRuntime( + entrypoint=ExplodingEntrypoint(), + runtime_adapter=StrategyRuntimeAdapter( + status_icon="📏", + required_feature_columns=frozenset(), + snapshot_date_columns=("as_of",), + max_snapshot_month_lag=1, + require_snapshot_manifest=False, + snapshot_contract_version=None, + managed_symbols_extractor=lambda *_args, **_kwargs: ("AAA", "BOXX"), + ), + runtime_settings=_build_runtime_settings(), + runtime_config={}, + merged_runtime_config={"safe_haven": "BOXX"}, + status_icon="📏", + logger=lambda _message: None, + ) + + monkeypatch.setattr( + strategy_runtime_module, + "load_feature_snapshot_guarded", + lambda path, **_kwargs: SimpleNamespace( + frame=[{"as_of": "2026-03-31", "symbol": "AAA"}], + metadata={ + "snapshot_guard_decision": "proceed", + "snapshot_as_of": "2026-03-31", + "snapshot_path": path, + "snapshot_age_days": 1, + }, + ), + ) + + result = runtime.evaluate( + ib=None, + current_holdings={"AAA"}, + historical_close_loader=lambda *_args, **_kwargs: None, + run_as_of=strategy_runtime_module.pd.Timestamp("2026-04-01"), + translator=lambda key, **_kwargs: key, + pacing_sec=0.5, + ) + + assert result.metadata["snapshot_guard_decision"] == "fail_closed" + assert "feature_snapshot_compute_failed:RuntimeError:boom" in result.metadata["fail_reason"] + assert result.decision.diagnostics["signal_description"] == "feature snapshot compute failed"