From 6e8cc058b821aea8a54015d4b39e02fbdd3dc198 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Mon, 6 Apr 2026 04:11:46 +0800 Subject: [PATCH] refactor shared strategy catalog contract --- src/quant_platform_kit/__init__.py | 30 ++- src/quant_platform_kit/common/strategies.py | 253 +++++++++++++++++++- tests/test_strategies.py | 104 ++++++++ 3 files changed, 385 insertions(+), 2 deletions(-) diff --git a/src/quant_platform_kit/__init__.py b/src/quant_platform_kit/__init__.py index 898df2f..812dfe8 100644 --- a/src/quant_platform_kit/__init__.py +++ b/src/quant_platform_kit/__init__.py @@ -1,6 +1,6 @@ """QuantPlatformKit public package surface.""" -__version__ = "0.5.0" +__version__ = "0.6.0" from .common.models import ( ExecutionReport, @@ -14,9 +14,23 @@ ) from .common.strategies import ( CRYPTO_DOMAIN, + PlatformStrategyPolicy, + StrategyCatalog, US_EQUITY_DOMAIN, StrategyDefinition, + StrategyMetadata, + build_platform_profile_matrix, + build_profile_aliases, + build_strategy_catalog, + build_strategy_index_rows, + get_catalog_compatible_platforms, + get_catalog_strategy_definition, + get_catalog_strategy_metadata, get_supported_profiles_for_platform, + get_enabled_profiles_for_platform, + normalize_profile_name, + resolve_catalog_profile, + resolve_platform_strategy_definition, resolve_strategy_definition, ) @@ -24,15 +38,29 @@ "__version__", "CRYPTO_DOMAIN", "ExecutionReport", + "PlatformStrategyPolicy", "OrderIntent", "PortfolioSnapshot", "Position", "PricePoint", "PriceSeries", "QuoteSnapshot", + "StrategyCatalog", "StrategyDefinition", "StrategyDecision", + "StrategyMetadata", "US_EQUITY_DOMAIN", + "build_platform_profile_matrix", + "build_profile_aliases", + "build_strategy_catalog", + "build_strategy_index_rows", + "get_catalog_compatible_platforms", + "get_catalog_strategy_definition", + "get_catalog_strategy_metadata", + "get_enabled_profiles_for_platform", "get_supported_profiles_for_platform", + "normalize_profile_name", + "resolve_catalog_profile", + "resolve_platform_strategy_definition", "resolve_strategy_definition", ] diff --git a/src/quant_platform_kit/common/strategies.py b/src/quant_platform_kit/common/strategies.py index c0c4a3d..8f4d2c9 100644 --- a/src/quant_platform_kit/common/strategies.py +++ b/src/quant_platform_kit/common/strategies.py @@ -3,7 +3,7 @@ from dataclasses import dataclass, field from importlib import import_module from types import ModuleType -from typing import Iterable +from typing import Iterable, Mapping US_EQUITY_DOMAIN = "us_equity" CRYPTO_DOMAIN = "crypto" @@ -23,6 +23,257 @@ class StrategyDefinition: components: tuple[StrategyComponentDefinition, ...] = field(default_factory=tuple) +@dataclass(frozen=True) +class StrategyMetadata: + canonical_profile: str + display_name: str + description: str + aliases: tuple[str, ...] = () + cadence: str | None = None + asset_scope: str | None = None + benchmark: str | None = None + role: str | None = None + status: str | None = None + + +@dataclass(frozen=True) +class StrategyCatalog: + definitions: Mapping[str, StrategyDefinition] + metadata: Mapping[str, StrategyMetadata] = field(default_factory=dict) + compatible_platforms: Mapping[str, frozenset[str]] = field(default_factory=dict) + profile_aliases: Mapping[str, str] = field(default_factory=dict) + + +@dataclass(frozen=True) +class PlatformStrategyPolicy: + platform_id: str + supported_domains: frozenset[str] + enabled_profiles: frozenset[str] + default_profile: str + rollback_profile: str + require_explicit_profile: bool = False + + +def normalize_profile_name(profile: str | None) -> str: + return str(profile or "").strip().lower() + + +def build_profile_aliases(metadata_map: Mapping[str, StrategyMetadata]) -> dict[str, str]: + aliases: dict[str, str] = {} + for canonical_profile, metadata in metadata_map.items(): + canonical = normalize_profile_name(canonical_profile) + if not canonical: + continue + for alias in metadata.aliases: + normalized_alias = normalize_profile_name(alias) + if not normalized_alias: + continue + existing = aliases.get(normalized_alias) + if existing is not None and existing != canonical: + raise ValueError( + f"Duplicate strategy alias {alias!r}; already assigned to {existing!r}" + ) + if normalized_alias in metadata_map and normalized_alias != canonical: + raise ValueError( + f"Strategy alias {alias!r} collides with canonical profile {normalized_alias!r}" + ) + aliases[normalized_alias] = canonical + return aliases + + +def build_strategy_catalog( + *, + strategy_definitions: Mapping[str, StrategyDefinition], + metadata: Mapping[str, StrategyMetadata] | None = None, + 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()} + metadata_map = { + normalize_profile_name(profile): value for profile, value in (metadata or {}).items() + } + compatibility_map = { + normalize_profile_name(profile): frozenset(platforms) + for profile, platforms in (compatible_platforms or {}).items() + } + missing_metadata = sorted(set(metadata_map) - set(definitions)) + if missing_metadata: + raise ValueError(f"Metadata provided for unknown profiles: {', '.join(missing_metadata)}") + missing_compatibility = sorted(set(compatibility_map) - set(definitions)) + if missing_compatibility: + raise ValueError( + f"Compatibility provided for unknown profiles: {', '.join(missing_compatibility)}" + ) + 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() + ) + } + return StrategyCatalog( + definitions=definitions, + metadata=metadata_map, + compatible_platforms=compatibility_map, + profile_aliases=aliases, + ) + + +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( + f"Unknown strategy profile={profile!r}; supported canonical values: {supported_text}; aliases: {alias_text}" + ) + + +def resolve_catalog_profile(profile: str | None, *, strategy_catalog: StrategyCatalog) -> str: + normalized = normalize_profile_name(profile) + if not normalized: + return normalized + return str(strategy_catalog.profile_aliases.get(normalized, normalized)) + + +def get_catalog_strategy_definition( + strategy_catalog: StrategyCatalog, + profile: str, +) -> StrategyDefinition: + canonical = resolve_catalog_profile(profile, strategy_catalog=strategy_catalog) + definition = strategy_catalog.definitions.get(canonical) + if definition is None: + raise _unsupported_profile_error( + profile=profile, + supported=strategy_catalog.definitions, + aliases=strategy_catalog.profile_aliases, + ) + return definition + + +def get_catalog_strategy_metadata( + strategy_catalog: StrategyCatalog, + profile: str, +) -> StrategyMetadata: + canonical = resolve_catalog_profile(profile, strategy_catalog=strategy_catalog) + metadata = strategy_catalog.metadata.get(canonical) + if metadata is None: + raise _unsupported_profile_error( + profile=profile, + supported=strategy_catalog.metadata, + aliases=strategy_catalog.profile_aliases, + ) + return metadata + + +def get_catalog_compatible_platforms( + strategy_catalog: StrategyCatalog, + profile: str, +) -> frozenset[str]: + canonical = resolve_catalog_profile(profile, strategy_catalog=strategy_catalog) + platforms = strategy_catalog.compatible_platforms.get(canonical) + if platforms is not None: + return frozenset(platforms) + definition = strategy_catalog.definitions.get(canonical) + if definition is None: + raise _unsupported_profile_error( + profile=profile, + supported=strategy_catalog.definitions, + aliases=strategy_catalog.profile_aliases, + ) + return definition.supported_platforms + + +def build_strategy_index_rows(strategy_catalog: StrategyCatalog) -> list[dict[str, object]]: + rows: list[dict[str, object]] = [] + for canonical_profile in sorted(strategy_catalog.definitions): + definition = strategy_catalog.definitions[canonical_profile] + metadata = strategy_catalog.metadata.get(canonical_profile) + rows.append( + { + "canonical_profile": canonical_profile, + "display_name": metadata.display_name if metadata else canonical_profile, + "aliases": metadata.aliases if metadata else (), + "description": metadata.description if metadata else "", + "cadence": metadata.cadence if metadata else None, + "asset_scope": metadata.asset_scope if metadata else None, + "benchmark": metadata.benchmark if metadata else None, + "role": metadata.role if metadata else None, + "status": metadata.status if metadata else None, + "component_names": tuple(component.name for component in definition.components), + "compatible_platforms": get_catalog_compatible_platforms( + strategy_catalog, + canonical_profile, + ), + } + ) + return rows + + +def get_enabled_profiles_for_platform( + platform_id: str, + *, + policy: PlatformStrategyPolicy, +) -> frozenset[str]: + if platform_id != policy.platform_id: + return frozenset() + return policy.enabled_profiles + + +def build_platform_profile_matrix( + strategy_catalog: StrategyCatalog, + *, + policy: PlatformStrategyPolicy, +) -> list[dict[str, object]]: + rows: list[dict[str, object]] = [] + for profile in sorted(policy.enabled_profiles): + definition = get_catalog_strategy_definition(strategy_catalog, profile) + metadata = strategy_catalog.metadata.get(definition.profile) + rows.append( + { + "platform": policy.platform_id, + "canonical_profile": definition.profile, + "display_name": metadata.display_name if metadata else definition.profile, + "aliases": metadata.aliases if metadata else (), + "enabled": True, + "is_default": definition.profile == policy.default_profile, + "is_rollback": definition.profile == policy.rollback_profile, + "domain": definition.domain, + } + ) + return rows + + +def resolve_platform_strategy_definition( + raw_value: str | None, + *, + platform_id: str, + strategy_catalog: StrategyCatalog, + policy: PlatformStrategyPolicy, +) -> StrategyDefinition: + if platform_id != policy.platform_id: + raise ValueError(f"Unsupported platform_id={platform_id!r}") + + normalized = normalize_profile_name(raw_value) + if policy.require_explicit_profile and not normalized: + raise EnvironmentError("STRATEGY_PROFILE is required") + + candidate = normalized or normalize_profile_name(policy.default_profile) + if not candidate: + raise EnvironmentError("STRATEGY_PROFILE is required") + + canonical = resolve_catalog_profile(candidate, strategy_catalog=strategy_catalog) + supported = ", ".join(sorted(policy.enabled_profiles)) or "" + if canonical not in policy.enabled_profiles: + raise ValueError( + f"Unsupported STRATEGY_PROFILE={raw_value!r}; supported values: {supported}" + ) + + definition = get_catalog_strategy_definition(strategy_catalog, canonical) + if definition.domain not in policy.supported_domains: + raise ValueError( + f"Unsupported strategy domain {definition.domain!r} for platform {platform_id!r}" + ) + return definition + + def get_strategy_component_map( definition: StrategyDefinition, ) -> dict[str, StrategyComponentDefinition]: diff --git a/tests/test_strategies.py b/tests/test_strategies.py index cb07ace..a5560bd 100644 --- a/tests/test_strategies.py +++ b/tests/test_strategies.py @@ -4,10 +4,22 @@ from quant_platform_kit.common.strategies import ( CRYPTO_DOMAIN, + PlatformStrategyPolicy, US_EQUITY_DOMAIN, StrategyComponentDefinition, StrategyDefinition, + StrategyMetadata, + build_platform_profile_matrix, + build_profile_aliases, + build_strategy_catalog, + build_strategy_index_rows, + get_catalog_compatible_platforms, + 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, ) @@ -102,3 +114,95 @@ def test_load_strategy_component_module_rejects_unknown_component(self) -> None: definition, component_name="allocation", ) + + def test_catalog_helpers_resolve_alias_and_metadata(self) -> None: + catalog = build_strategy_catalog( + strategy_definitions=self.strategy_definitions, + metadata={ + "global_etf_rotation": StrategyMetadata( + canonical_profile="global_etf_rotation", + display_name="Global ETF Rotation Defense", + description="rotation", + aliases=("global_macro_etf_rotation",), + cadence="quarterly", + benchmark="VOO", + ) + }, + compatible_platforms={ + "global_etf_rotation": frozenset({"ibkr"}), + }, + ) + + self.assertEqual( + resolve_catalog_profile("global_macro_etf_rotation", strategy_catalog=catalog), + "global_etf_rotation", + ) + self.assertEqual( + get_catalog_strategy_definition(catalog, "global_macro_etf_rotation").profile, + "global_etf_rotation", + ) + self.assertEqual( + get_catalog_strategy_metadata(catalog, "global_etf_rotation").display_name, + "Global ETF Rotation Defense", + ) + self.assertEqual( + get_catalog_compatible_platforms(catalog, "global_etf_rotation"), + frozenset({"ibkr"}), + ) + rows = build_strategy_index_rows(catalog) + by_profile = {row["canonical_profile"]: row for row in rows} + self.assertEqual(by_profile["global_etf_rotation"]["display_name"], "Global ETF Rotation Defense") + + def test_platform_policy_helpers_build_matrix_and_resolve_enabled_profile(self) -> None: + catalog = build_strategy_catalog( + strategy_definitions=self.strategy_definitions, + metadata={ + "global_etf_rotation": StrategyMetadata( + canonical_profile="global_etf_rotation", + display_name="Global ETF Rotation Defense", + description="rotation", + aliases=("global_macro_etf_rotation",), + ), + }, + ) + 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", + require_explicit_profile=True, + ) + + self.assertEqual( + get_enabled_profiles_for_platform("ibkr", policy=policy), + frozenset({"global_etf_rotation"}), + ) + matrix = build_platform_profile_matrix(catalog, policy=policy) + self.assertEqual(matrix[0]["display_name"], "Global ETF Rotation Defense") + definition = resolve_platform_strategy_definition( + "global_macro_etf_rotation", + platform_id="ibkr", + strategy_catalog=catalog, + policy=policy, + ) + self.assertEqual(definition.profile, "global_etf_rotation") + + def test_build_profile_aliases_rejects_duplicate_alias(self) -> None: + with self.assertRaisesRegex(ValueError, "Duplicate strategy alias"): + build_profile_aliases( + { + "one": StrategyMetadata( + canonical_profile="one", + display_name="One", + description="", + aliases=("shared",), + ), + "two": StrategyMetadata( + canonical_profile="two", + display_name="Two", + description="", + aliases=("shared",), + ), + } + )