From f1df03b8ff1598e222e9f26cca3d9b714cea21a6 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 7 Apr 2026 02:39:21 +0800 Subject: [PATCH] refactor: finalize shared strategy contract cleanup --- docs/next_window_cleanup_split.md | 383 ++++++++++++++++++ docs/strategy_contract_migration.md | 139 +++++++ src/quant_platform_kit/__init__.py | 38 +- src/quant_platform_kit/common/strategies.py | 240 +++++++++-- .../common/strategy_contracts.py | 198 +++++++++ src/quant_platform_kit/strategy_contracts.py | 29 ++ tests/test_strategies.py | 57 ++- tests/test_strategy_contracts.py | 243 +++++++++++ 8 files changed, 1258 insertions(+), 69 deletions(-) create mode 100644 docs/next_window_cleanup_split.md create mode 100644 docs/strategy_contract_migration.md create mode 100644 src/quant_platform_kit/common/strategy_contracts.py create mode 100644 src/quant_platform_kit/strategy_contracts.py create mode 100644 tests/test_strategy_contracts.py diff --git a/docs/next_window_cleanup_split.md b/docs/next_window_cleanup_split.md new file mode 100644 index 0000000..d845e9f --- /dev/null +++ b/docs/next_window_cleanup_split.md @@ -0,0 +1,383 @@ +# Next-window cleanup split plan + +_Verified snapshot: 2026-04-07_ + +This document pre-splits the **next compatibility-window cleanup** into small PR-sized batches. + +It is a planning document only. +Do **not** merge these removals early. +The current window still intentionally keeps rollback hooks. + +## Scope + +Remove only migration-window compatibility layers that were kept during PR1-PR6. + +Do **not** use this cleanup window to: + +- change any strategy formulas +- change thresholds or risk semantics +- refactor broker integrations beyond the boundary-removal work +- mix feature work into cleanup PRs + +## Common release gates + +Every removal PR below should wait until all of these are true: + +1. the unified-entrypoint mainline has shipped for at least one stable release window +2. rollback to the previous release tag is still possible during the cutover +3. regression tests for the affected repo are green on the unified path +4. `rg` shows the legacy symbol is only referenced in the files being deleted or updated in that PR +5. any research or test-only imports have been moved off the compatibility shim first + +## Recommended PR order + +1. `NW1` QuantPlatformKit deprecated helper removal - done +2. `NW2` LongBridgePlatform allocation-loader removal - done +3. `NW3` CharlesSchwabPlatform allocation-loader removal - done +4. `NW4` UsEquityStrategies metadata lift for IB cleanup - done +5. `NW5` InteractiveBrokersPlatform legacy-module bridge removal - done +6. `NW6` BinancePlatform direct component-loader and main-wrapper removal - done + +This order keeps the easy and low-coupling deletions first, and leaves the IB / Binance boundary cleanup for after their remaining test and adapter dependencies are gone. + +## Execution status + +- Completed on the current branch: `NW1`, `NW2`, `NW3`, `NW4`, `NW5`, `NW6` +- Remaining planned batch after that: none + +--- + +## NW1 - QuantPlatformKit: remove deprecated global compatibility helpers + +**Status:** completed on the current branch. + +### Goal + +Delete the migration-window helpers that were superseded by `StrategyCatalog` + `PlatformStrategyPolicy`. + +### Repositories + +- `QuantPlatformKit` + +### Delete + +- `src/quant_platform_kit/common/strategies.py` + - `get_supported_profiles_for_platform(...)` + - `resolve_strategy_definition(...)` + - the deprecation warning strings and legacy call path that support them +- `src/quant_platform_kit/__init__.py` + - legacy exports for the two deprecated helpers +- tests and docs that still assert the deprecated flow + +### Keep + +- `resolve_platform_strategy_definition(...)` +- `get_enabled_profiles_for_platform(...)` +- `build_platform_profile_matrix(...)` +- platform-local `strategy_registry.py` wrappers that already sit on top of the new policy/catalog API + +### Current references to clean up + +- `src/quant_platform_kit/common/strategies.py` +- `src/quant_platform_kit/__init__.py` +- `tests/test_strategies.py` +- `tests/test_strategy_contracts.py` +- `docs/strategy_contract_migration.md` + +### Done when + +- `rg -n "resolve_strategy_definition\(|get_supported_profiles_for_platform\(" .` in `QuantPlatformKit` only finds historical notes that were intentionally rewritten, or finds no runtime code at all +- tests cover only the catalog/policy path + +### Suggested validation + +```bash +cd /Users/lisiyi/Projects/QuantPlatformKit +python3 -m ruff check src tests +PYTHONPATH=src python3 -m unittest tests.test_strategy_contracts tests.test_strategies tests.test_models -v +``` + +--- + +## NW2 - LongBridgePlatform: remove allocation compatibility loader + +**Status:** completed on the current branch. + +### Goal + +Delete the last platform-local allocation shim API now that `main.py` already runs through `strategy_runtime` + unified entrypoint. + +### Repositories + +- `LongBridgePlatform` + +### Delete + +- `strategy_loader.py` + - `load_allocation_module(...)` +- tests that still verify allocation-module loading + +### Current references to clean up + +- `strategy_loader.py` +- `tests/test_strategy_loader.py` + +### Preconditions + +None beyond the common release gates. +The mainline path is already on `load_strategy_entrypoint_for_profile(...)`. + +### Done when + +- `rg -n "load_allocation_module\(" .` in `LongBridgePlatform` returns no matches +- loader tests only assert unified entrypoint loading + +### Suggested validation + +```bash +cd /Users/lisiyi/Projects/LongBridgePlatform +python3 -m ruff check strategy_loader.py tests/test_strategy_loader.py tests/test_strategy_runtime.py +PYTHONPATH=.:/Users/lisiyi/Projects/QuantPlatformKit/src:/Users/lisiyi/Projects/UsEquityStrategies/src python -m unittest \ + tests.test_strategy_loader tests.test_strategy_runtime tests.test_rebalance_service tests.test_request_handling -v +``` + +--- + +## NW3 - CharlesSchwabPlatform: remove allocation compatibility loader + +**Status:** completed on the current branch. + +### Goal + +Do the same cleanup as LongBridge, but in the Schwab runtime. + +### Repositories + +- `CharlesSchwabPlatform` + +### Delete + +- `strategy_loader.py` + - `load_allocation_module(...)` +- tests that still verify allocation-module loading + +### Current references to clean up + +- `strategy_loader.py` +- `tests/test_strategy_loader.py` + +### Preconditions + +None beyond the common release gates. +The mainline path is already on `load_strategy_entrypoint_for_profile(...)`. + +### Done when + +- `rg -n "load_allocation_module\(" .` in `CharlesSchwabPlatform` returns no matches +- loader tests only assert unified entrypoint loading + +### Suggested validation + +```bash +cd /Users/lisiyi/Projects/CharlesSchwabPlatform +python3 -m ruff check strategy_loader.py tests/test_strategy_loader.py tests/test_strategy_runtime.py +PYTHONPATH=.:/Users/lisiyi/Projects/QuantPlatformKit/src:/Users/lisiyi/Projects/UsEquityStrategies/src python -m unittest \ + tests.test_strategy_loader tests.test_strategy_runtime tests.test_rebalance_service tests.test_request_handling -v +``` + +--- + +## NW4 - UsEquityStrategies: lift the remaining IB-facing legacy metadata + +**Status:** completed on the current branch. A typed runtime adapter now exists in `UsEquityStrategies`, and `InteractiveBrokersPlatform` now consumes it without the legacy signal-module fallback. + +### Goal + +Move the last **strategy-side** metadata that IB still reads through the legacy signal module into the unified side, so the platform can delete the legacy bridge in the next PR. + +### Repositories + +- `UsEquityStrategies` +- possibly `QuantPlatformKit` only if the existing manifest/default-config shape proves insufficient + +### Strategy-side data still consumed through the IB legacy module today + +`InteractiveBrokersPlatform/strategy_runtime.py` still reads these through `legacy_module`: + +- `STATUS_ICON` +- `load_runtime_parameters(...)` +- `REQUIRED_FEATURE_COLUMNS` +- `SNAPSHOT_DATE_COLUMNS` +- `MAX_SNAPSHOT_MONTH_LAG` +- `REQUIRE_SNAPSHOT_MANIFEST` +- `SNAPSHOT_CONTRACT_VERSION` +- `extract_managed_symbols(...)` + +### Recommended direction + +Prefer one of these, in this order: + +1. move the data into `manifest.default_config` when it is declarative +2. expose a strategy-side adapter object next to the unified entrypoint for IB-only snapshot/runtime metadata +3. only extend QuantPlatformKit contract types if the first two options are clearly not enough + +Do **not** reintroduce platform-shaped return payloads into `StrategyDecision`. + +### Done when + +- IB runtime can build snapshot guards and managed-symbol metadata without importing the legacy signal module +- the strategy-side contract remains formula-neutral +- there is a regression test in `UsEquityStrategies` for the new metadata path + +### Suggested validation + +```bash +cd /Users/lisiyi/Projects/UsEquityStrategies +python3 -m ruff check src tests +PYTHONPATH=src:/Users/lisiyi/Projects/QuantPlatformKit/src python -m unittest tests.test_entrypoints tests.test_catalog -v +``` + +--- + +## NW5 - InteractiveBrokersPlatform: remove the legacy signal-module bridge + +### Goal + +Delete the remaining platform-side bridge to legacy signal modules after NW4 has moved the required metadata to the unified side. + +### Repositories + +- `InteractiveBrokersPlatform` +- depends on `UsEquityStrategies` work from `NW4` + +### Delete + +- `strategy_loader.py` + - `load_signal_logic_module(...)` +- `strategy_runtime.py` + - `legacy_module` field on `LoadedStrategyRuntime` + - all `getattr(..., legacy_module, ...)` snapshot/runtime fallbacks + - direct `load_runtime_parameters(...)` call through the legacy module +- `main.py` + - `STRATEGY_LOGIC` + - `strategy_check_sma` + - `strategy_compute_13612w_momentum` + - `strategy_compute_signals` (currently dead) + - any wrapper logic that only exists to proxy legacy module helpers +- tests that still exercise `load_signal_logic_module(...)` + +### Current references to clean up + +- `strategy_loader.py` +- `strategy_runtime.py` +- `main.py` +- `tests/test_strategy_loader.py` + +### Additional notes + +`compute_signals(...)` stays. +It is the platform mainline entry into `STRATEGY_RUNTIME.evaluate(...)` and `decision_mapper.map_strategy_decision(...)`. +Only the legacy module bridge should go away. + +### Done when + +- `rg -n "load_signal_logic_module\(|\bSTRATEGY_LOGIC\b" .` in `InteractiveBrokersPlatform` returns no runtime matches +- snapshot guard configuration comes from the unified side only +- IB smoke/regression tests still pass with no behavior drift + +### Suggested validation + +```bash +cd /Users/lisiyi/Projects/InteractiveBrokersPlatform +python3 -m ruff check main.py strategy_loader.py strategy_runtime.py tests/test_strategy_loader.py tests/test_strategy_runtime.py +PYTHONPATH=.:/Users/lisiyi/Projects/QuantPlatformKit/src:/Users/lisiyi/Projects/UsEquityStrategies/src ./.venv/bin/python -m pytest \ + tests/test_strategy_loader.py \ + tests/test_strategy_runtime.py \ + tests/test_decision_mapper.py \ + tests/test_rebalance_service.py \ + tests/test_request_handling.py \ + tests/test_runtime_config_support.py -q +``` + +--- + +## NW6 - BinancePlatform: remove direct core/rotation component loading and main-level compatibility wrappers + +**Status:** completed on the current branch. + +### Goal + +Finish the platform-side cleanup so Binance only loads the unified entrypoint and decision mappers, while research/tests use either unified runtime outputs or explicit upstream strategy imports. + +### Repositories + +- `BinancePlatform` +- optional follow-up in `CryptoStrategies` only if an extra typed helper is still missing + +### Delete from platform runtime code + +- `strategy_loader.py` + - `load_strategy_component(...)` +- `strategy_runtime.py` + - `core_module` + - `rotation_module` + - temporary helper methods: + - `compute_allocation_budgets(...)` + - `get_dynamic_btc_base_order(...)` + - `allocate_trend_buy_budget(...)` + - `refresh_rotation_pool(...)` +- `main.py` + - `get_dynamic_btc_target_ratio(...)` + - `get_dynamic_btc_base_order(...)` + - `rank_normalize(...)` + - `build_stable_quality_pool(...)` + - `refresh_rotation_pool(...)` + - `select_rotation_weights(...)` + - `allocate_trend_buy_budget(...)` + +### Current references that must move first + +- `tests/test_strategy_loader.py` + - currently still asserts `load_strategy_component(...)` +- `tests/test_trend_pool_loading.py` + - currently patches `main.build_stable_quality_pool(...)` and calls `main.refresh_rotation_pool(...)` +- `research/backtest.py` + - still calls `get_dynamic_btc_base_order(...)` +- any remaining tests that import main-level compatibility wrappers instead of using unified runtime output or explicit `crypto_strategies` imports + +### Preferred migration targets + +- use `StrategyDecision.diagnostics` for runtime-facing BTC / trend-rotation facts +- import explicit upstream modules from `CryptoStrategies` in research-only code when a helper is intentionally research-only +- keep platform runtime code on `load_strategy_entrypoint_for_profile(...)` only + +### Done when + +- `rg -n "load_strategy_component\(|get_dynamic_btc_target_ratio\(|get_dynamic_btc_base_order\(|build_stable_quality_pool\(|refresh_rotation_pool\(|select_rotation_weights\(|allocate_trend_buy_budget\(" .` in `BinancePlatform` only finds intended research-side upstream imports or returns no matches for platform runtime code +- `strategy_runtime.py` loads the unified entrypoint only +- loader tests no longer assert direct `core` / `rotation` module resolution + +### Suggested validation + +```bash +cd /Users/lisiyi/Projects/BinancePlatform +python3 -m ruff check main.py strategy_loader.py strategy_runtime.py research/backtest.py tests +PYTHONPATH=.:/Users/lisiyi/Projects/QuantPlatformKit/src:/Users/lisiyi/Projects/CryptoStrategies/src python -m unittest discover -s tests -p 'test_*.py' -v +``` + +--- + +## Cross-repo grep checklist before opening removal PRs + +Use these quick checks before starting each deletion batch: + +```bash +for repo in QuantPlatformKit InteractiveBrokersPlatform LongBridgePlatform CharlesSchwabPlatform BinancePlatform; do + echo "===== $repo =====" + cd /Users/lisiyi/Projects/$repo || exit 1 + rg -n "resolve_strategy_definition\(|get_supported_profiles_for_platform\(|load_signal_logic_module\(|STRATEGY_LOGIC\b|load_allocation_module\(|load_strategy_component\(|get_dynamic_btc_target_ratio\(|get_dynamic_btc_base_order\(|build_stable_quality_pool\(|refresh_rotation_pool\(|select_rotation_weights\(|allocate_trend_buy_budget\(" . --glob '!**/__pycache__/**' + echo + done +``` + +If that output still shows runtime or test references outside the PR scope, split again before deleting. diff --git a/docs/strategy_contract_migration.md b/docs/strategy_contract_migration.md new file mode 100644 index 0000000..5653c7b --- /dev/null +++ b/docs/strategy_contract_migration.md @@ -0,0 +1,139 @@ +# Strategy contract migration notes + +## What changed in PR1 + +QuantPlatformKit now provides a shared strategy contract for the platform/strategy split: + +- `quant_platform_kit.strategy_contracts.StrategyManifest` +- `quant_platform_kit.strategy_contracts.StrategyContext` +- `quant_platform_kit.strategy_contracts.StrategyDecision` +- `quant_platform_kit.strategy_contracts.StrategyEntrypoint` +- `quant_platform_kit.strategy_contracts.StrategyRuntimeAdapter` +- `quant_platform_kit.common.strategies.load_strategy_entrypoint(...)` +- `quant_platform_kit.common.strategies.validate_strategy_manifest(...)` +- `quant_platform_kit.common.strategies.validate_strategy_decision(...)` +- `quant_platform_kit.common.strategy_contracts.validate_strategy_runtime_adapter(...)` + +PR1 kept the old global compatibility helpers for one compatibility window. +They have now been removed in the next-window cleanup batch; platforms should use +`resolve_platform_strategy_definition(...)` and `get_enabled_profiles_for_platform(...)`. + +## How strategy repos should migrate next + +### 1. Add manifest + evaluate entrypoint per live profile + +Each profile should expose either: + +- an explicit `entrypoint` object, or +- a module-level `manifest` + `evaluate(ctx)` pair. + +Recommended import target for downstream platforms: + +```python +from quant_platform_kit.common.strategies import load_strategy_entrypoint +``` + +### 2. Keep legacy component modules during the migration window + +`load_strategy_entrypoint(...)` resolves candidates in this order: + +1. `StrategyDefinition.entrypoint` +2. a catalog component named `entrypoint` +3. existing legacy component modules that now expose `entrypoint`, or `manifest` + `evaluate` + +That means strategy repos can add the new contract without deleting legacy `signal_logic`, `allocation`, `core`, or `rotation` modules in the same PR. + +### 3. Move platform-only fields out of strategy returns + +The new `StrategyDecision` only keeps: + +- `positions` +- `budgets` +- `risk_flags` +- `diagnostics` + +Fields such as `sell_order_symbols`, `buy_order_symbols`, `portfolio_rows`, `cash_sweep_symbol`, and other broker/UI ordering hints should move to platform-side decision mappers in later PRs. + +### 4. Move platform-side runtime metadata behind a typed adapter when needed + +Some platforms still need strategy-owned runtime metadata during the migration window. + +For IBKR feature-snapshot profiles, that metadata now moves behind: + +- `StrategyRuntimeAdapter` +- strategy-repo getter functions such as `get_platform_runtime_adapter(profile, platform_id=...)` + +Use this adapter for strategy-owned runtime metadata such as: + +- status icon defaults +- required feature snapshot columns +- snapshot contract / manifest expectations +- managed-symbol extraction helpers +- temporary runtime-config loaders kept during the migration window + +Do not move broker execution fields back into `StrategyDecision`. + +## Platform migration expectations + +Platform repos should gradually switch from: + +- `resolve_strategy_definition(...)` +- `load_strategy_component_module(...)` + +to: + +- `resolve_platform_strategy_definition(...)` +- `load_strategy_entrypoint(...)` +- platform-local decision mappers + +During the compatibility window, old component loaders may stay in place behind feature flags or rollback paths. + +## Cross-repo follow-up + +- `UsEquityStrategies`: add per-profile manifest + entrypoint adapters for every live profile. +- `CryptoStrategies`: add `crypto_leader_rotation` entrypoint and artifact/input manifest declarations. +- `InteractiveBrokersPlatform`: stop reading strategy private constants in `main.py`, consume unified decisions via a mapper. +- `LongBridgePlatform` / `CharlesSchwabPlatform`: remove allocation shims and hard-coded strategy asset lists. +- `BinancePlatform`: replace `core` / `rotation` shims with a unified entrypoint and explicit artifact contract. + +## PR6 cleanup status + +### Mainline status by repository + +| Repository | Mainline execution path | Compatibility window still kept | +| --- | --- | --- | +| `InteractiveBrokersPlatform` | `main.py` -> `strategy_runtime.load_strategy_runtime(...)` -> unified entrypoint -> `decision_mapper.map_strategy_decision(...)` | no legacy signal-module bridge remains. | +| `LongBridgePlatform` | `main.py` -> `strategy_runtime.load_strategy_runtime(...)` -> unified entrypoint -> `decision_mapper.map_strategy_decision_to_plan(...)` | no platform-local allocation loader remains. | +| `CharlesSchwabPlatform` | `main.py` -> `strategy_runtime.load_strategy_runtime(...)` -> unified entrypoint -> `decision_mapper.map_strategy_decision_to_plan(...)` | no platform-local allocation loader remains. | +| `BinancePlatform` | `main.py` / cycle services -> `strategy_runtime.load_strategy_runtime(...)` -> unified entrypoint -> decision mappers | no direct component loader or main/runtime compatibility wrapper remains. | + +### Cleanup completed in the current window + +- Strategy repositories now expose manifest + entrypoint adapters for live profiles without changing trading formulas. +- `UsEquityStrategies` now also exposes a typed IBKR runtime adapter for migration-window snapshot/runtime metadata. +- Platform-only output fields remain outside `StrategyDecision`; platforms now derive orders and notifications through local mappers. +- `BinancePlatform` removed repo-local `strategy_core.py` and `strategy/rotation.py`; live-pool loading now follows an explicit artifact contract instead of assuming a sibling checkout is the only source. +- `LongBridgePlatform` and `CharlesSchwabPlatform` mainline flows no longer pass long chains of strategy-specific allocation arguments through `main.py`. +- `QuantPlatformKit` removed the deprecated global resolver helpers after one full compatibility window. +- `LongBridgePlatform` and `CharlesSchwabPlatform` removed `load_allocation_module(...)`; loaders now expose unified entrypoints only. +- `InteractiveBrokersPlatform` removed `load_signal_logic_module(...)`, `STRATEGY_LOGIC`, and the legacy snapshot/runtime fallback path; runtime metadata now comes from `UsEquityStrategies.get_platform_runtime_adapter(...)` only. +- `BinancePlatform` removed `load_strategy_component(...)`, direct `core` / `rotation` loading, and the temporary main/runtime wrapper helpers; runtime and execution now stay on unified entrypoints plus decision mappers only. + +### Compatibility APIs intentionally kept for one more window + +No platform-local compatibility API from PR1-PR6 is intentionally kept after the current cleanup batches. + +### Next-window removals + +Remove these only after the next platform release has shipped on the unified entrypoint path and rollback is no longer needed: + +No remaining platform-runtime cleanup item is pending from this migration track. + +Detailed repo-by-repo split: [`next_window_cleanup_split.md`](./next_window_cleanup_split.md) + +### Cross-repo dependency notes + +- Platform repos depend on `QuantPlatformKit` for contract types, loader validation, and shared broker helpers. +- `InteractiveBrokersPlatform`, `LongBridgePlatform`, and `CharlesSchwabPlatform` depend on `UsEquityStrategies` entrypoints staying API-compatible for at least one window. +- `BinancePlatform` depends on `CryptoStrategies` entrypoints plus the explicit live-pool artifact contract; sibling repo checkout is only an optional fallback now. +- `InteractiveBrokersPlatform` now depends on `UsEquityStrategies.get_platform_runtime_adapter(...)` for runtime metadata on the unified path. diff --git a/src/quant_platform_kit/__init__.py b/src/quant_platform_kit/__init__.py index 812dfe8..c3843c4 100644 --- a/src/quant_platform_kit/__init__.py +++ b/src/quant_platform_kit/__init__.py @@ -12,55 +12,81 @@ QuoteSnapshot, StrategyDecision, ) +from .common.strategy_contracts import ( + BudgetIntent, + PositionTarget, + StrategyContext, + StrategyContractValidationError, + StrategyDecision as StrategyContractDecision, + StrategyEntrypoint, + StrategyManifest, + StrategyRuntimeAdapter, + validate_strategy_decision, + validate_strategy_manifest, + validate_strategy_runtime_adapter, +) from .common.strategies import ( CRYPTO_DOMAIN, PlatformStrategyPolicy, StrategyCatalog, - US_EQUITY_DOMAIN, StrategyDefinition, + StrategyEntrypointDefinition, StrategyMetadata, + US_EQUITY_DOMAIN, build_platform_profile_matrix, build_profile_aliases, build_strategy_catalog, build_strategy_index_rows, + build_strategy_manifest, get_catalog_compatible_platforms, get_catalog_strategy_definition, get_catalog_strategy_metadata, - get_supported_profiles_for_platform, get_enabled_profiles_for_platform, + load_strategy_entrypoint, normalize_profile_name, resolve_catalog_profile, resolve_platform_strategy_definition, - resolve_strategy_definition, ) __all__ = [ "__version__", + "BudgetIntent", "CRYPTO_DOMAIN", "ExecutionReport", "PlatformStrategyPolicy", "OrderIntent", "PortfolioSnapshot", "Position", + "PositionTarget", "PricePoint", "PriceSeries", "QuoteSnapshot", "StrategyCatalog", - "StrategyDefinition", + "StrategyContext", + "StrategyContractDecision", + "StrategyContractValidationError", "StrategyDecision", + "StrategyDefinition", + "StrategyEntrypoint", + "StrategyEntrypointDefinition", + "StrategyManifest", "StrategyMetadata", + "StrategyRuntimeAdapter", "US_EQUITY_DOMAIN", "build_platform_profile_matrix", "build_profile_aliases", "build_strategy_catalog", "build_strategy_index_rows", + "build_strategy_manifest", "get_catalog_compatible_platforms", "get_catalog_strategy_definition", "get_catalog_strategy_metadata", "get_enabled_profiles_for_platform", - "get_supported_profiles_for_platform", + "load_strategy_entrypoint", "normalize_profile_name", "resolve_catalog_profile", "resolve_platform_strategy_definition", - "resolve_strategy_definition", + "validate_strategy_decision", + "validate_strategy_manifest", + "validate_strategy_runtime_adapter", ] diff --git a/src/quant_platform_kit/common/strategies.py b/src/quant_platform_kit/common/strategies.py index 8f4d2c9..37434d9 100644 --- a/src/quant_platform_kit/common/strategies.py +++ b/src/quant_platform_kit/common/strategies.py @@ -3,7 +3,15 @@ from dataclasses import dataclass, field from importlib import import_module from types import ModuleType -from typing import Iterable, Mapping +from typing import Any, Iterable, Mapping + +from .strategy_contracts import ( + CallableStrategyEntrypoint, + StrategyContractValidationError, + StrategyEntrypoint, + StrategyManifest, + validate_strategy_manifest, +) US_EQUITY_DOMAIN = "us_equity" CRYPTO_DOMAIN = "crypto" @@ -15,12 +23,22 @@ class StrategyComponentDefinition: module_path: str +@dataclass(frozen=True) +class StrategyEntrypointDefinition: + module_path: str + attribute_name: str = "entrypoint" + + @dataclass(frozen=True) class StrategyDefinition: profile: str domain: str supported_platforms: frozenset[str] components: tuple[StrategyComponentDefinition, ...] = field(default_factory=tuple) + entrypoint: StrategyEntrypointDefinition | None = None + required_inputs: frozenset[str] = frozenset() + compatible_capabilities: frozenset[str] = frozenset() + default_config: Mapping[str, Any] = field(default_factory=dict) @dataclass(frozen=True) @@ -88,7 +106,10 @@ def build_strategy_catalog( compatible_platforms: Mapping[str, frozenset[str]] | None = None, profile_aliases: Mapping[str, str] | None = None, ) -> StrategyCatalog: - definitions = {normalize_profile_name(profile): definition for profile, definition in strategy_definitions.items()} + definitions = { + normalize_profile_name(profile): definition + for profile, definition in strategy_definitions.items() + } metadata_map = { normalize_profile_name(profile): value for profile, value in (metadata or {}).items() } @@ -107,7 +128,9 @@ def build_strategy_catalog( aliases = { normalize_profile_name(alias): normalize_profile_name(canonical) for alias, canonical in ( - profile_aliases.items() if profile_aliases is not None else build_profile_aliases(metadata_map).items() + profile_aliases.items() + if profile_aliases is not None + else build_profile_aliases(metadata_map).items() ) } return StrategyCatalog( @@ -118,7 +141,12 @@ def build_strategy_catalog( ) -def _unsupported_profile_error(*, profile: str | None, supported: Iterable[str], aliases: Iterable[str]) -> ValueError: +def _unsupported_profile_error( + *, + profile: str | None, + supported: Iterable[str], + aliases: Iterable[str], +) -> ValueError: supported_text = ", ".join(sorted(supported)) or "" alias_text = ", ".join(sorted(aliases)) or "" return ValueError( @@ -202,6 +230,10 @@ def build_strategy_index_rows(strategy_catalog: StrategyCatalog) -> list[dict[st strategy_catalog, canonical_profile, ), + "has_entrypoint": definition.entrypoint is not None + or "entrypoint" in {component.name for component in definition.components}, + "required_inputs": definition.required_inputs, + "compatible_capabilities": definition.compatible_capabilities, } ) return rows @@ -310,52 +342,178 @@ def load_strategy_component_modules( } -def get_supported_profiles_for_platform( - strategy_definitions: dict[str, StrategyDefinition], - platform_supported_domains: dict[str, frozenset[str]], +def build_strategy_manifest( + definition: StrategyDefinition, *, - platform_id: str, -) -> frozenset[str]: - return frozenset( - profile - for profile, definition in strategy_definitions.items() - if platform_id in definition.supported_platforms - and definition.domain in platform_supported_domains.get(platform_id, frozenset()) + metadata: StrategyMetadata | None = None, +) -> StrategyManifest: + manifest = StrategyManifest( + profile=definition.profile, + domain=definition.domain, + display_name=(metadata.display_name if metadata else definition.profile.replace("_", " ").title()), + description=( + metadata.description + if metadata + else f"Legacy entrypoint adapter for strategy profile {definition.profile}." + ), + aliases=metadata.aliases if metadata else (), + required_inputs=definition.required_inputs, + compatible_capabilities=definition.compatible_capabilities, + default_config=dict(definition.default_config), ) + return validate_strategy_manifest(manifest) -def resolve_strategy_definition( - raw_value: str | None, +def _coerce_entrypoint_candidate( + candidate: object, *, - platform_id: str, - strategy_definitions: dict[str, StrategyDefinition], - platform_supported_domains: dict[str, frozenset[str]], - default_profile: str | None = None, - require_explicit: bool = False, -) -> StrategyDefinition: - if require_explicit and not str(raw_value or "").strip(): - raise EnvironmentError("STRATEGY_PROFILE is required") + source: str, +) -> StrategyEntrypoint: + manifest = getattr(candidate, "manifest", None) + evaluate = getattr(candidate, "evaluate", None) + if manifest is None or not callable(evaluate): + raise StrategyContractValidationError( + f"{source} must expose manifest and callable evaluate(ctx)" + ) + return CallableStrategyEntrypoint( + manifest=validate_strategy_manifest(manifest), + _evaluate=evaluate, + ) - profile = (raw_value or default_profile or "").strip().lower() - if not profile: - raise EnvironmentError("STRATEGY_PROFILE is required") - supported_profiles = get_supported_profiles_for_platform( - strategy_definitions, - platform_supported_domains, - platform_id=platform_id, - ) - supported = ", ".join(sorted(supported_profiles)) +def _module_entrypoint_candidate( + module: ModuleType, + *, + module_ref: StrategyEntrypointDefinition, + definition: StrategyDefinition, + metadata: StrategyMetadata | None, +) -> StrategyEntrypoint | None: + attribute_name = module_ref.attribute_name.strip() + if attribute_name: + explicit = getattr(module, attribute_name, None) + if explicit is not None: + return _coerce_entrypoint_candidate( + explicit, + source=f"{module.__name__}.{attribute_name}", + ) + + factory = getattr(module, "build_entrypoint", None) + if callable(factory): + return _coerce_entrypoint_candidate(factory(), source=f"{module.__name__}.build_entrypoint()") + + evaluate = getattr(module, "evaluate", None) + if not callable(evaluate): + return None + + manifest = getattr(module, "manifest", None) + if manifest is None: + manifest = build_strategy_manifest(definition, metadata=metadata) + else: + validate_strategy_manifest(manifest) + return CallableStrategyEntrypoint(manifest=manifest, _evaluate=evaluate) + + +def _iter_entrypoint_candidates( + definition: StrategyDefinition, +) -> tuple[StrategyEntrypointDefinition, ...]: + candidates: list[StrategyEntrypointDefinition] = [] + seen: set[tuple[str, str]] = set() + + def append_candidate(module_path: str, attribute_name: str = "entrypoint") -> None: + key = (module_path, attribute_name) + if key in seen: + return + seen.add(key) + candidates.append( + StrategyEntrypointDefinition(module_path=module_path, attribute_name=attribute_name) + ) - definition = strategy_definitions.get(profile) - if definition is None or platform_id not in definition.supported_platforms: - raise ValueError( - f"Unsupported STRATEGY_PROFILE={raw_value!r}; supported values: {supported}" + if definition.entrypoint is not None: + append_candidate( + definition.entrypoint.module_path, + definition.entrypoint.attribute_name, ) - if definition.domain not in platform_supported_domains.get(platform_id, frozenset()): - raise ValueError( - f"Unsupported strategy domain {definition.domain!r} for platform {platform_id!r}" + component_map = get_strategy_component_map(definition) + entrypoint_component = component_map.get("entrypoint") + if entrypoint_component is not None: + append_candidate(entrypoint_component.module_path) + + for component in definition.components: + append_candidate(component.module_path) + + return tuple(candidates) + + +def _validate_entrypoint_compatibility( + entrypoint: StrategyEntrypoint, + definition: StrategyDefinition, + *, + platform_id: str | None, + available_inputs: Iterable[str] | None, + available_capabilities: Iterable[str] | None, +) -> None: + manifest = validate_strategy_manifest(entrypoint.manifest) + + if available_inputs is not None and manifest.required_inputs: + missing_inputs = manifest.required_inputs - frozenset(available_inputs) + if missing_inputs: + raise StrategyContractValidationError( + "Strategy manifest requires missing inputs: " + f"{', '.join(sorted(missing_inputs))}" + ) + + if manifest.compatible_capabilities: + capabilities = frozenset(available_capabilities or ()) + missing_capabilities = manifest.compatible_capabilities - capabilities + if missing_capabilities: + raise StrategyContractValidationError( + "Strategy manifest requires missing capabilities: " + f"{', '.join(sorted(missing_capabilities))}" + ) + return + + if platform_id is not None and platform_id not in definition.supported_platforms: + supported = ", ".join(sorted(definition.supported_platforms)) or "" + raise StrategyContractValidationError( + f"Strategy profile {definition.profile!r} is not compatible with platform " + f"{platform_id!r}; supported_platforms={supported}" ) - return definition + +def load_strategy_entrypoint( + definition: StrategyDefinition, + *, + metadata: StrategyMetadata | None = None, + platform_id: str | None = None, + available_inputs: Iterable[str] | None = None, + available_capabilities: Iterable[str] | None = None, +) -> StrategyEntrypoint: + candidates = _iter_entrypoint_candidates(definition) + if not candidates: + raise ValueError(f"Strategy profile {definition.profile!r} has no entrypoint candidates") + + for module_ref in candidates: + module = import_module(module_ref.module_path) + entrypoint = _module_entrypoint_candidate( + module, + module_ref=module_ref, + definition=definition, + metadata=metadata, + ) + if entrypoint is None: + continue + _validate_entrypoint_compatibility( + entrypoint, + definition, + platform_id=platform_id, + available_inputs=available_inputs, + available_capabilities=available_capabilities, + ) + return entrypoint + + candidate_modules = ", ".join(module_ref.module_path for module_ref in candidates) + raise ValueError( + f"Strategy profile {definition.profile!r} does not expose a unified entrypoint; " + f"checked modules: {candidate_modules}" + ) diff --git a/src/quant_platform_kit/common/strategy_contracts.py b/src/quant_platform_kit/common/strategy_contracts.py new file mode 100644 index 0000000..5e74e03 --- /dev/null +++ b/src/quant_platform_kit/common/strategy_contracts.py @@ -0,0 +1,198 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Callable, Mapping, Protocol +import math + + +class StrategyContractValidationError(ValueError): + """Raised when a strategy manifest or decision violates the shared contract.""" + + +@dataclass(frozen=True) +class StrategyManifest: + profile: str + domain: str + display_name: str + description: str + aliases: tuple[str, ...] = () + required_inputs: frozenset[str] = frozenset() + compatible_capabilities: frozenset[str] = frozenset() + default_config: Mapping[str, Any] = field(default_factory=dict) + + +@dataclass(frozen=True) +class PositionTarget: + symbol: str + target_weight: float | None = None + target_value: float | None = None + role: str | None = None + order_preference: str | None = None + + +@dataclass(frozen=True) +class BudgetIntent: + name: str + symbol: str | None = None + amount: float | None = None + unit: str = "quote_ccy" + purpose: str | None = None + + +@dataclass(frozen=True) +class StrategyContext: + as_of: Any + market_data: Mapping[str, Any] = field(default_factory=dict) + portfolio: Any | None = None + state: Mapping[str, Any] = field(default_factory=dict) + runtime_config: Mapping[str, Any] = field(default_factory=dict) + capabilities: Mapping[str, Any] = field(default_factory=dict) + artifacts: Mapping[str, Any] = field(default_factory=dict) + + +@dataclass(frozen=True) +class StrategyDecision: + positions: tuple[PositionTarget, ...] = () + budgets: tuple[BudgetIntent, ...] = () + risk_flags: tuple[str, ...] = () + diagnostics: Mapping[str, Any] = field(default_factory=dict) + + +class StrategyEntrypoint(Protocol): + manifest: StrategyManifest + + def evaluate(self, ctx: StrategyContext) -> StrategyDecision: ... + + +@dataclass(frozen=True) +class StrategyRuntimeAdapter: + status_icon: str = "🐤" + required_feature_columns: frozenset[str] = frozenset() + snapshot_date_columns: tuple[str, ...] = ("as_of", "snapshot_date") + max_snapshot_month_lag: int = 1 + require_snapshot_manifest: bool = False + snapshot_contract_version: str | None = None + runtime_parameter_loader: Callable[..., Mapping[str, object]] | None = None + managed_symbols_extractor: Callable[..., tuple[str, ...]] | None = None + + +@dataclass(frozen=True) +class CallableStrategyEntrypoint: + manifest: StrategyManifest + _evaluate: Callable[[StrategyContext], StrategyDecision] + + def evaluate(self, ctx: StrategyContext) -> StrategyDecision: + decision = self._evaluate(ctx) + return validate_strategy_decision(decision) + + +def _ensure_non_empty_string(value: str, *, field_name: str) -> None: + if not isinstance(value, str) or not value.strip(): + raise StrategyContractValidationError(f"{field_name} must be a non-empty string") + + +def _ensure_string_set(values: frozenset[str] | tuple[str, ...], *, field_name: str) -> None: + for value in values: + _ensure_non_empty_string(value, field_name=field_name) + + +def _ensure_finite_number(value: float | None, *, field_name: str) -> None: + if value is None: + return + if not isinstance(value, (int, float)) or not math.isfinite(float(value)): + raise StrategyContractValidationError(f"{field_name} must be a finite number when provided") + + +def validate_strategy_manifest(manifest: StrategyManifest) -> StrategyManifest: + if not isinstance(manifest, StrategyManifest): + raise StrategyContractValidationError( + f"manifest must be StrategyManifest, got {type(manifest).__name__}" + ) + + _ensure_non_empty_string(manifest.profile, field_name="manifest.profile") + _ensure_non_empty_string(manifest.domain, field_name="manifest.domain") + _ensure_non_empty_string(manifest.display_name, field_name="manifest.display_name") + _ensure_non_empty_string(manifest.description, field_name="manifest.description") + _ensure_string_set(manifest.aliases, field_name="manifest.aliases[]") + _ensure_string_set(manifest.required_inputs, field_name="manifest.required_inputs[]") + _ensure_string_set( + manifest.compatible_capabilities, + field_name="manifest.compatible_capabilities[]", + ) + if not isinstance(manifest.default_config, Mapping): + raise StrategyContractValidationError("manifest.default_config must be a mapping") + return manifest + + +def validate_strategy_decision(decision: StrategyDecision) -> StrategyDecision: + if not isinstance(decision, StrategyDecision): + raise StrategyContractValidationError( + f"decision must be StrategyDecision, got {type(decision).__name__}" + ) + + if not isinstance(decision.diagnostics, Mapping): + raise StrategyContractValidationError("decision.diagnostics must be a mapping") + + for position in decision.positions: + if not isinstance(position, PositionTarget): + raise StrategyContractValidationError( + f"decision.positions entries must be PositionTarget, got {type(position).__name__}" + ) + _ensure_non_empty_string(position.symbol, field_name="position.symbol") + if position.target_weight is None and position.target_value is None: + raise StrategyContractValidationError( + f"position {position.symbol!r} must set target_weight or target_value" + ) + _ensure_finite_number(position.target_weight, field_name="position.target_weight") + _ensure_finite_number(position.target_value, field_name="position.target_value") + + for budget in decision.budgets: + if not isinstance(budget, BudgetIntent): + raise StrategyContractValidationError( + f"decision.budgets entries must be BudgetIntent, got {type(budget).__name__}" + ) + _ensure_non_empty_string(budget.name, field_name="budget.name") + if budget.symbol is not None: + _ensure_non_empty_string(budget.symbol, field_name="budget.symbol") + _ensure_non_empty_string(budget.unit, field_name="budget.unit") + _ensure_finite_number(budget.amount, field_name="budget.amount") + + for risk_flag in decision.risk_flags: + _ensure_non_empty_string(risk_flag, field_name="decision.risk_flags[]") + + return decision + + +def validate_strategy_runtime_adapter(adapter: StrategyRuntimeAdapter) -> StrategyRuntimeAdapter: + if not isinstance(adapter, StrategyRuntimeAdapter): + raise StrategyContractValidationError( + f"runtime adapter must be StrategyRuntimeAdapter, got {type(adapter).__name__}" + ) + + _ensure_non_empty_string(adapter.status_icon, field_name="runtime_adapter.status_icon") + _ensure_string_set( + adapter.required_feature_columns, + field_name="runtime_adapter.required_feature_columns[]", + ) + _ensure_string_set( + adapter.snapshot_date_columns, + field_name="runtime_adapter.snapshot_date_columns[]", + ) + if not isinstance(adapter.max_snapshot_month_lag, int) or adapter.max_snapshot_month_lag < 0: + raise StrategyContractValidationError( + "runtime_adapter.max_snapshot_month_lag must be a non-negative integer" + ) + if adapter.snapshot_contract_version is not None: + _ensure_non_empty_string( + adapter.snapshot_contract_version, + field_name="runtime_adapter.snapshot_contract_version", + ) + if adapter.runtime_parameter_loader is not None and not callable(adapter.runtime_parameter_loader): + raise StrategyContractValidationError( + "runtime_adapter.runtime_parameter_loader must be callable when provided" + ) + if adapter.managed_symbols_extractor is not None and not callable(adapter.managed_symbols_extractor): + raise StrategyContractValidationError( + "runtime_adapter.managed_symbols_extractor must be callable when provided" + ) + return adapter diff --git a/src/quant_platform_kit/strategy_contracts.py b/src/quant_platform_kit/strategy_contracts.py new file mode 100644 index 0000000..c68cff3 --- /dev/null +++ b/src/quant_platform_kit/strategy_contracts.py @@ -0,0 +1,29 @@ +from .common.strategy_contracts import ( + BudgetIntent, + CallableStrategyEntrypoint, + PositionTarget, + StrategyContext, + StrategyContractValidationError, + StrategyDecision, + StrategyEntrypoint, + StrategyManifest, + StrategyRuntimeAdapter, + validate_strategy_decision, + validate_strategy_manifest, + validate_strategy_runtime_adapter, +) + +__all__ = [ + "BudgetIntent", + "CallableStrategyEntrypoint", + "PositionTarget", + "StrategyContext", + "StrategyContractValidationError", + "StrategyDecision", + "StrategyEntrypoint", + "StrategyManifest", + "StrategyRuntimeAdapter", + "validate_strategy_decision", + "validate_strategy_manifest", + "validate_strategy_runtime_adapter", +] diff --git a/tests/test_strategies.py b/tests/test_strategies.py index a5560bd..1b36d13 100644 --- a/tests/test_strategies.py +++ b/tests/test_strategies.py @@ -17,11 +17,9 @@ get_catalog_strategy_definition, get_catalog_strategy_metadata, get_enabled_profiles_for_platform, - get_supported_profiles_for_platform, resolve_catalog_profile, - resolve_platform_strategy_definition, load_strategy_component_module, - resolve_strategy_definition, + resolve_platform_strategy_definition, ) @@ -55,45 +53,60 @@ def setUp(self) -> None: "ibkr": frozenset({US_EQUITY_DOMAIN}), "binance": frozenset({CRYPTO_DOMAIN}), } - - def test_get_supported_profiles_for_platform_filters_by_domain_and_platform(self) -> None: - supported = get_supported_profiles_for_platform( - self.strategy_definitions, - self.platform_supported_domains, + self.strategy_catalog = build_strategy_catalog( + strategy_definitions=self.strategy_definitions, + ) + self.ibkr_policy = PlatformStrategyPolicy( platform_id="ibkr", + supported_domains=self.platform_supported_domains["ibkr"], + enabled_profiles=frozenset({"global_etf_rotation"}), + default_profile="global_etf_rotation", + rollback_profile="global_etf_rotation", + require_explicit_profile=True, + ) + self.binance_policy = PlatformStrategyPolicy( + platform_id="binance", + supported_domains=self.platform_supported_domains["binance"], + enabled_profiles=frozenset({"crypto_leader_rotation"}), + default_profile="crypto_leader_rotation", + rollback_profile="crypto_leader_rotation", + ) + + def test_get_enabled_profiles_for_platform_reads_platform_policy(self) -> None: + supported = get_enabled_profiles_for_platform( + "ibkr", + policy=self.ibkr_policy, ) self.assertEqual(supported, frozenset({"global_etf_rotation"})) - def test_resolve_strategy_definition_uses_default_profile_when_allowed(self) -> None: - definition = resolve_strategy_definition( + def test_resolve_platform_strategy_definition_uses_default_profile_when_allowed(self) -> None: + definition = resolve_platform_strategy_definition( None, platform_id="binance", - strategy_definitions=self.strategy_definitions, - platform_supported_domains=self.platform_supported_domains, - default_profile="crypto_leader_rotation", + strategy_catalog=self.strategy_catalog, + policy=self.binance_policy, ) self.assertEqual(definition.profile, "crypto_leader_rotation") self.assertEqual(definition.domain, CRYPTO_DOMAIN) - def test_resolve_strategy_definition_requires_explicit_when_requested(self) -> None: + def test_resolve_platform_strategy_definition_requires_explicit_when_requested(self) -> None: with self.assertRaisesRegex(EnvironmentError, "STRATEGY_PROFILE is required"): - resolve_strategy_definition( + resolve_platform_strategy_definition( None, platform_id="ibkr", - strategy_definitions=self.strategy_definitions, - platform_supported_domains=self.platform_supported_domains, - require_explicit=True, + strategy_catalog=self.strategy_catalog, + policy=self.ibkr_policy, ) - def test_resolve_strategy_definition_rejects_profile_outside_platform_domain(self) -> None: + def test_resolve_platform_strategy_definition_rejects_profile_outside_platform_domain(self) -> None: with self.assertRaisesRegex(ValueError, "Unsupported STRATEGY_PROFILE"): - resolve_strategy_definition( + resolve_platform_strategy_definition( "crypto_leader_rotation", platform_id="ibkr", - strategy_definitions=self.strategy_definitions, - platform_supported_domains=self.platform_supported_domains, + strategy_catalog=self.strategy_catalog, + policy=self.ibkr_policy, ) def test_load_strategy_component_module_imports_named_component(self) -> None: diff --git a/tests/test_strategy_contracts.py b/tests/test_strategy_contracts.py new file mode 100644 index 0000000..ff9e326 --- /dev/null +++ b/tests/test_strategy_contracts.py @@ -0,0 +1,243 @@ +from __future__ import annotations + +from types import ModuleType +import sys +import unittest + +from quant_platform_kit.common.strategies import ( + CRYPTO_DOMAIN, + PlatformStrategyPolicy, + StrategyComponentDefinition, + StrategyDefinition, + StrategyEntrypointDefinition, + StrategyMetadata, + US_EQUITY_DOMAIN, + build_strategy_manifest, + build_strategy_catalog, + get_enabled_profiles_for_platform, + load_strategy_entrypoint, + resolve_platform_strategy_definition, +) +from quant_platform_kit.strategy_contracts import ( + CallableStrategyEntrypoint, + PositionTarget, + StrategyContext, + StrategyContractValidationError, + StrategyDecision, + StrategyManifest, + StrategyRuntimeAdapter, + validate_strategy_decision, + validate_strategy_manifest, + validate_strategy_runtime_adapter, +) + + +class StrategyContractMigrationTests(unittest.TestCase): + def _install_module(self, name: str, **attrs: object) -> None: + module = ModuleType(name) + for key, value in attrs.items(): + setattr(module, key, value) + sys.modules[name] = module + self.addCleanup(sys.modules.pop, name, None) + + def test_build_strategy_manifest_uses_metadata_and_new_contract_fields(self) -> None: + definition = StrategyDefinition( + profile="global_etf_rotation", + domain=US_EQUITY_DOMAIN, + supported_platforms=frozenset({"ibkr"}), + required_inputs=frozenset({"market_data", "portfolio"}), + compatible_capabilities=frozenset({"rebalance_orders"}), + default_config={"safe_haven": "BIL"}, + ) + metadata = StrategyMetadata( + canonical_profile="global_etf_rotation", + display_name="Global ETF Rotation Defense", + description="legacy adapter", + aliases=("global_macro_etf_rotation",), + ) + + manifest = build_strategy_manifest(definition, metadata=metadata) + + self.assertEqual(manifest.display_name, "Global ETF Rotation Defense") + self.assertEqual(manifest.aliases, ("global_macro_etf_rotation",)) + self.assertEqual(manifest.required_inputs, frozenset({"market_data", "portfolio"})) + self.assertEqual(manifest.default_config["safe_haven"], "BIL") + + def test_load_strategy_entrypoint_prefers_explicit_entrypoint_definition(self) -> None: + module_name = "_quant_platform_kit_test_explicit_entrypoint" + entrypoint = CallableStrategyEntrypoint( + manifest=StrategyManifest( + profile="global_etf_rotation", + domain=US_EQUITY_DOMAIN, + display_name="Global ETF Rotation Defense", + description="explicit entrypoint", + required_inputs=frozenset({"market_data"}), + compatible_capabilities=frozenset({"rebalance_orders"}), + ), + _evaluate=lambda ctx: StrategyDecision( + positions=(PositionTarget(symbol="SPY", target_weight=1.0),), + diagnostics={"as_of": ctx.as_of}, + ), + ) + self._install_module(module_name, entrypoint=entrypoint) + definition = StrategyDefinition( + profile="global_etf_rotation", + domain=US_EQUITY_DOMAIN, + supported_platforms=frozenset({"ibkr"}), + entrypoint=StrategyEntrypointDefinition(module_path=module_name), + ) + + loaded = load_strategy_entrypoint( + definition, + platform_id="ibkr", + available_inputs={"market_data"}, + available_capabilities={"rebalance_orders", "notifications"}, + ) + decision = loaded.evaluate(StrategyContext(as_of="2026-04-06")) + + self.assertEqual(loaded.manifest.profile, "global_etf_rotation") + self.assertEqual(decision.positions[0].symbol, "SPY") + self.assertEqual(decision.diagnostics["as_of"], "2026-04-06") + + def test_load_strategy_entrypoint_falls_back_to_legacy_component_module(self) -> None: + module_name = "_quant_platform_kit_test_legacy_component" + manifest = StrategyManifest( + profile="tech_pullback_cash_buffer", + domain=US_EQUITY_DOMAIN, + display_name="Tech Pullback Cash Buffer", + description="legacy component with manifest/evaluate", + required_inputs=frozenset({"market_data"}), + ) + + def evaluate(ctx: StrategyContext) -> StrategyDecision: + self.assertEqual(ctx.as_of, "2026-04-06") + return StrategyDecision( + positions=(PositionTarget(symbol="BOXX", target_weight=0.25),), + risk_flags=("cash_buffer",), + ) + + self._install_module(module_name, manifest=manifest, evaluate=evaluate) + definition = StrategyDefinition( + profile="tech_pullback_cash_buffer", + domain=US_EQUITY_DOMAIN, + supported_platforms=frozenset({"ibkr"}), + components=( + StrategyComponentDefinition( + name="signal_logic", + module_path=module_name, + ), + ), + ) + + loaded = load_strategy_entrypoint( + definition, + platform_id="ibkr", + available_inputs={"market_data"}, + ) + decision = loaded.evaluate(StrategyContext(as_of="2026-04-06")) + + self.assertEqual(decision.risk_flags, ("cash_buffer",)) + self.assertEqual(loaded.manifest.display_name, "Tech Pullback Cash Buffer") + + def test_load_strategy_entrypoint_rejects_missing_inputs_and_legacy_platform_fallback(self) -> None: + module_name = "_quant_platform_kit_test_requirements" + + def evaluate(_ctx: StrategyContext) -> StrategyDecision: + return StrategyDecision( + positions=(PositionTarget(symbol="BTCUSDT", target_weight=1.0),), + ) + + self._install_module(module_name, evaluate=evaluate) + definition = StrategyDefinition( + profile="crypto_leader_rotation", + domain=CRYPTO_DOMAIN, + supported_platforms=frozenset({"binance"}), + components=( + StrategyComponentDefinition(name="core", module_path=module_name), + ), + required_inputs=frozenset({"artifacts"}), + ) + metadata = StrategyMetadata( + canonical_profile="crypto_leader_rotation", + display_name="Crypto Leader Rotation", + description="legacy fallback manifest", + ) + + with self.assertRaisesRegex(StrategyContractValidationError, "missing inputs"): + load_strategy_entrypoint( + definition, + metadata=metadata, + platform_id="binance", + available_inputs={"market_data"}, + ) + + with self.assertRaisesRegex(StrategyContractValidationError, "not compatible with platform"): + load_strategy_entrypoint( + definition, + metadata=metadata, + platform_id="ibkr", + available_inputs={"artifacts"}, + ) + + def test_validators_reject_invalid_manifest_and_decision_shapes(self) -> None: + with self.assertRaisesRegex(StrategyContractValidationError, "manifest.display_name"): + validate_strategy_manifest( + StrategyManifest( + profile="global_etf_rotation", + domain=US_EQUITY_DOMAIN, + display_name="", + description="bad", + ) + ) + + with self.assertRaisesRegex(StrategyContractValidationError, "must set target_weight or target_value"): + validate_strategy_decision( + StrategyDecision( + positions=(PositionTarget(symbol="SPY"),), + ) + ) + + with self.assertRaisesRegex( + StrategyContractValidationError, + "runtime_adapter.max_snapshot_month_lag", + ): + validate_strategy_runtime_adapter( + StrategyRuntimeAdapter(max_snapshot_month_lag=-1) + ) + + adapter = validate_strategy_runtime_adapter( + StrategyRuntimeAdapter( + status_icon="🧲", + required_feature_columns=frozenset({"symbol", "close"}), + snapshot_contract_version="contract.v1", + runtime_parameter_loader=lambda **_kwargs: {"safe_haven": "BOXX"}, + managed_symbols_extractor=lambda *_args, **_kwargs: ("AAPL", "BOXX"), + ) + ) + self.assertEqual(adapter.status_icon, "🧲") + + def test_platform_policy_helpers_replace_legacy_global_helpers(self) -> None: + strategy_definitions = { + "global_etf_rotation": StrategyDefinition( + profile="global_etf_rotation", + domain=US_EQUITY_DOMAIN, + supported_platforms=frozenset({"ibkr"}), + ) + } + catalog = build_strategy_catalog(strategy_definitions=strategy_definitions) + policy = PlatformStrategyPolicy( + platform_id="ibkr", + supported_domains=frozenset({US_EQUITY_DOMAIN}), + enabled_profiles=frozenset({"global_etf_rotation"}), + default_profile="global_etf_rotation", + rollback_profile="global_etf_rotation", + ) + supported = get_enabled_profiles_for_platform("ibkr", policy=policy) + definition = resolve_platform_strategy_definition( + "global_etf_rotation", + platform_id="ibkr", + strategy_catalog=catalog, + policy=policy, + ) + self.assertEqual(supported, frozenset({"global_etf_rotation"})) + self.assertEqual(definition.profile, "global_etf_rotation")