Merge pull request #6 from QuantStrategyLab/codex/ues-shared-strategy-catalog

[codex] refactor shared strategy catalog access

Author: Pigbibi

pyproject.toml

@@ -9,7 +9,7 @@ description = "Shared US equity strategy catalog and implementations"
 readme = "README.md"
 requires-python = ">=3.11"
 dependencies = [
-  "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@v0.6.0",
+  "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@6e8cc058b821aea8a54015d4b39e02fbdd3dc198",
 ]
 
 [tool.setuptools]

src/us_equity_strategies/__init__.py

@@ -1,7 +1,9 @@
 __all__ = [
+    "STRATEGY_CATALOG",
     "STRATEGY_DEFINITIONS",
     "get_compatible_platforms",
     "get_profile_aliases",
+    "get_strategy_catalog",
     "get_strategy_index_rows",
     "get_strategy_definition",
     "get_strategy_definitions",
@@ -14,9 +16,11 @@
 
 def __getattr__(name: str):
     if name in {
+        "STRATEGY_CATALOG",
         "STRATEGY_DEFINITIONS",
         "get_profile_aliases",
         "get_compatible_platforms",
+        "get_strategy_catalog",
         "get_strategy_index_rows",
         "get_strategy_definition",
         "get_strategy_definitions",

src/us_equity_strategies/catalog.py

@@ -1,11 +1,18 @@
 from __future__ import annotations
 
-from dataclasses import dataclass
-
 from quant_platform_kit.common.strategies import (
+    StrategyCatalog,
     StrategyComponentDefinition,
     StrategyDefinition,
+    StrategyMetadata,
     US_EQUITY_DOMAIN,
+    build_strategy_catalog,
+    build_strategy_index_rows,
+    get_catalog_compatible_platforms,
+    get_catalog_strategy_definition,
+    get_catalog_strategy_metadata,
+    normalize_profile_name as qpk_normalize_profile_name,
+    resolve_catalog_profile,
 )
 
 GLOBAL_ETF_ROTATION_PROFILE = "global_etf_rotation"
@@ -23,20 +30,6 @@
     CASH_BUFFER_BRANCH_DEFAULT_PROFILE: frozenset({"ibkr"}),
 }
 
-
-@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
-
-
 # `supported_platforms` 仍保留为兼容镜像，避免一次性改动所有平台 runtime。
 # 平台真正的启用状态由各自 runtime 仓库维护；UES 这里只表达策略层兼容性。
 def _build_strategy_definition(
@@ -151,68 +144,45 @@ def _build_strategy_definition(
     for alias in metadata.aliases
 }
 
+STRATEGY_CATALOG: StrategyCatalog = build_strategy_catalog(
+    strategy_definitions=STRATEGY_DEFINITIONS,
+    metadata=STRATEGY_METADATA,
+    compatible_platforms=STRATEGY_PLATFORM_COMPATIBILITY,
+    profile_aliases=PROFILE_ALIASES,
+)
+
 
 def normalize_profile_name(profile: str | None) -> str:
-    return str(profile or "").strip().lower()
+    return qpk_normalize_profile_name(profile)
 
 
 def resolve_canonical_profile(profile: str | None) -> str:
-    normalized = normalize_profile_name(profile)
-    return PROFILE_ALIASES.get(normalized, normalized)
+    return resolve_catalog_profile(profile, strategy_catalog=STRATEGY_CATALOG)
 
 
 def get_strategy_definitions() -> dict[str, StrategyDefinition]:
     return dict(STRATEGY_DEFINITIONS)
 
 
+def get_strategy_catalog() -> StrategyCatalog:
+    return STRATEGY_CATALOG
+
+
 def get_strategy_platform_compatibility_map() -> dict[str, frozenset[str]]:
     return dict(STRATEGY_PLATFORM_COMPATIBILITY)
 
 
 def get_compatible_platforms(profile: str) -> frozenset[str]:
-    canonical = resolve_canonical_profile(profile)
-    if canonical not in STRATEGY_PLATFORM_COMPATIBILITY:
-        supported = ", ".join(sorted(STRATEGY_PLATFORM_COMPATIBILITY)) or "<none>"
-        aliases = ", ".join(sorted(PROFILE_ALIASES)) or "<none>"
-        raise ValueError(
-            f"Unknown us_equity strategy profile={profile!r}; supported canonical values: {supported}; aliases: {aliases}"
-        )
-    return STRATEGY_PLATFORM_COMPATIBILITY[canonical]
+    return get_catalog_compatible_platforms(STRATEGY_CATALOG, profile)
 
 
 def get_strategy_definition(profile: str) -> StrategyDefinition:
-    canonical = resolve_canonical_profile(profile)
-    if canonical not in STRATEGY_DEFINITIONS:
-        supported = ", ".join(sorted(STRATEGY_DEFINITIONS)) or "<none>"
-        aliases = ", ".join(sorted(PROFILE_ALIASES)) or "<none>"
-        raise ValueError(
-            f"Unknown us_equity strategy profile={profile!r}; supported canonical values: {supported}; aliases: {aliases}"
-        )
-    return STRATEGY_DEFINITIONS[canonical]
+    return get_catalog_strategy_definition(STRATEGY_CATALOG, profile)
 
 
 
 def get_strategy_index_rows() -> list[dict[str, object]]:
-    rows: list[dict[str, object]] = []
-    for canonical_profile in sorted(STRATEGY_METADATA):
-        metadata = STRATEGY_METADATA[canonical_profile]
-        definition = STRATEGY_DEFINITIONS[canonical_profile]
-        rows.append(
-            {
-                "canonical_profile": metadata.canonical_profile,
-                "display_name": metadata.display_name,
-                "aliases": metadata.aliases,
-                "description": metadata.description,
-                "cadence": metadata.cadence,
-                "asset_scope": metadata.asset_scope,
-                "benchmark": metadata.benchmark,
-                "role": metadata.role,
-                "status": metadata.status,
-                "component_names": tuple(component.name for component in definition.components),
-                "compatible_platforms": STRATEGY_PLATFORM_COMPATIBILITY[canonical_profile],
-            }
-        )
-    return rows
+    return build_strategy_index_rows(STRATEGY_CATALOG)
 
 
 
@@ -221,14 +191,7 @@ def get_strategy_metadata_map() -> dict[str, StrategyMetadata]:
 
 
 def get_strategy_metadata(profile: str) -> StrategyMetadata:
-    canonical = resolve_canonical_profile(profile)
-    if canonical not in STRATEGY_METADATA:
-        supported = ", ".join(sorted(STRATEGY_METADATA)) or "<none>"
-        aliases = ", ".join(sorted(PROFILE_ALIASES)) or "<none>"
-        raise ValueError(
-            f"Unknown us_equity strategy profile={profile!r}; supported canonical values: {supported}; aliases: {aliases}"
-        )
-    return STRATEGY_METADATA[canonical]
+    return get_catalog_strategy_metadata(STRATEGY_CATALOG, profile)
 
 
 def get_profile_aliases() -> dict[str, str]:

src/us_equity_strategies/platform_registry_support.py

@@ -1,23 +1,32 @@
 from __future__ import annotations
 
-from quant_platform_kit.common.strategies import StrategyDefinition
-
-from .catalog import (
-    get_strategy_definition,
-    get_strategy_metadata,
-    resolve_canonical_profile,
+from quant_platform_kit.common.strategies import (
+    PlatformStrategyPolicy,
+    StrategyDefinition,
+    build_platform_profile_matrix as qpk_build_platform_profile_matrix,
+    get_enabled_profiles_for_platform as qpk_get_enabled_profiles_for_platform,
+    resolve_platform_strategy_definition as qpk_resolve_platform_strategy_definition,
 )
 
+from .catalog import STRATEGY_CATALOG
+
 
 def get_enabled_profiles_for_platform(
     platform_id: str,
     *,
     expected_platform_id: str,
     enabled_profiles: frozenset[str],
 ) -> frozenset[str]:
-    if platform_id != expected_platform_id:
-        return frozenset()
-    return enabled_profiles
+    return qpk_get_enabled_profiles_for_platform(
+        platform_id,
+        policy=PlatformStrategyPolicy(
+            platform_id=expected_platform_id,
+            supported_domains=frozenset(),
+            enabled_profiles=enabled_profiles,
+            default_profile="",
+            rollback_profile="",
+        ),
+    )
 
 
 def build_platform_profile_matrix(
@@ -27,23 +36,16 @@ def build_platform_profile_matrix(
     default_profile: str,
     rollback_profile: str,
 ) -> list[dict[str, object]]:
-    rows: list[dict[str, object]] = []
-    for profile in sorted(enabled_profiles):
-        definition = get_strategy_definition(profile)
-        metadata = get_strategy_metadata(profile)
-        rows.append(
-            {
-                "platform": platform_id,
-                "canonical_profile": definition.profile,
-                "display_name": metadata.display_name,
-                "aliases": metadata.aliases,
-                "enabled": True,
-                "is_default": definition.profile == default_profile,
-                "is_rollback": definition.profile == rollback_profile,
-                "domain": definition.domain,
-            }
-        )
-    return rows
+    return qpk_build_platform_profile_matrix(
+        STRATEGY_CATALOG,
+        policy=PlatformStrategyPolicy(
+            platform_id=platform_id,
+            supported_domains=frozenset(),
+            enabled_profiles=enabled_profiles,
+            default_profile=default_profile,
+            rollback_profile=rollback_profile,
+        ),
+    )
 
 
 def resolve_platform_strategy_definition(
@@ -56,29 +58,16 @@ def resolve_platform_strategy_definition(
     default_profile: str | None = None,
     require_explicit: bool = False,
 ) -> StrategyDefinition:
-    if platform_id != expected_platform_id:
-        raise ValueError(f"Unsupported platform_id={platform_id!r}")
-
-    normalized = str(raw_value or "").strip()
-    if require_explicit and not normalized:
-        raise EnvironmentError("STRATEGY_PROFILE is required")
-
-    candidate = normalized or str(default_profile or "").strip()
-    if not candidate:
-        raise EnvironmentError("STRATEGY_PROFILE is required")
-
-    canonical_profile = resolve_canonical_profile(candidate)
-    supported = ", ".join(sorted(enabled_profiles))
-
-    if canonical_profile not in enabled_profiles:
-        raise ValueError(
-            f"Unsupported STRATEGY_PROFILE={raw_value!r}; supported values: {supported}"
-        )
-
-    definition = get_strategy_definition(canonical_profile)
-    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}"
-        )
-
-    return definition
+    return qpk_resolve_platform_strategy_definition(
+        raw_value,
+        platform_id=platform_id,
+        strategy_catalog=STRATEGY_CATALOG,
+        policy=PlatformStrategyPolicy(
+            platform_id=expected_platform_id,
+            supported_domains=platform_supported_domains.get(expected_platform_id, frozenset()),
+            enabled_profiles=enabled_profiles,
+            default_profile=default_profile or "",
+            rollback_profile=default_profile or "",
+            require_explicit_profile=require_explicit,
+        ),
+    )