Add shared strategy component loader

Author: Pigbibi

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "quant-platform-kit"
-version = "0.5.0"
+version = "0.6.0"
 description = "Shared broker adapters, domain models, execution ports, and notification utilities for QuantStrategyLab strategies."
 readme = "README.md"
 requires-python = ">=3.9"

src/quant_platform_kit/common/strategies.py

@@ -1,16 +1,62 @@
 from __future__ import annotations
 
-from dataclasses import dataclass
+from dataclasses import dataclass, field
+from importlib import import_module
+from types import ModuleType
+from typing import Iterable
 
 US_EQUITY_DOMAIN = "us_equity"
 CRYPTO_DOMAIN = "crypto"
 
 
+@dataclass(frozen=True)
+class StrategyComponentDefinition:
+    name: str
+    module_path: str
+
+
 @dataclass(frozen=True)
 class StrategyDefinition:
     profile: str
     domain: str
     supported_platforms: frozenset[str]
+    components: tuple[StrategyComponentDefinition, ...] = field(default_factory=tuple)
+
+
+def get_strategy_component_map(
+    definition: StrategyDefinition,
+) -> dict[str, StrategyComponentDefinition]:
+    return {component.name: component for component in definition.components}
+
+
+def load_strategy_component_module(
+    definition: StrategyDefinition,
+    *,
+    component_name: str,
+) -> ModuleType:
+    component_map = get_strategy_component_map(definition)
+    component = component_map.get(component_name)
+    if component is None:
+        available = ", ".join(sorted(component_map)) or "<none>"
+        raise ValueError(
+            f"Strategy profile {definition.profile!r} does not expose component "
+            f"{component_name!r}; available components: {available}"
+        )
+    return import_module(component.module_path)
+
+
+def load_strategy_component_modules(
+    definition: StrategyDefinition,
+    *,
+    component_names: Iterable[str],
+) -> dict[str, ModuleType]:
+    return {
+        component_name: load_strategy_component_module(
+            definition,
+            component_name=component_name,
+        )
+        for component_name in component_names
+    }
 
 
 def get_supported_profiles_for_platform(

tests/test_strategies.py

@@ -5,8 +5,10 @@
 from quant_platform_kit.common.strategies import (
     CRYPTO_DOMAIN,
     US_EQUITY_DOMAIN,
+    StrategyComponentDefinition,
     StrategyDefinition,
     get_supported_profiles_for_platform,
+    load_strategy_component_module,
     resolve_strategy_definition,
 )
 
@@ -18,11 +20,23 @@ def setUp(self) -> None:
                 profile="global_etf_rotation",
                 domain=US_EQUITY_DOMAIN,
                 supported_platforms=frozenset({"ibkr", "schwab", "longbridge"}),
+                components=(
+                    StrategyComponentDefinition(
+                        name="signal_logic",
+                        module_path="math",
+                    ),
+                ),
             ),
             "crypto_leader_rotation": StrategyDefinition(
                 profile="crypto_leader_rotation",
                 domain=CRYPTO_DOMAIN,
                 supported_platforms=frozenset({"binance"}),
+                components=(
+                    StrategyComponentDefinition(
+                        name="core",
+                        module_path="math",
+                    ),
+                ),
             ),
         }
         self.platform_supported_domains = {
@@ -69,3 +83,22 @@ def test_resolve_strategy_definition_rejects_profile_outside_platform_domain(sel
                 strategy_definitions=self.strategy_definitions,
                 platform_supported_domains=self.platform_supported_domains,
             )
+
+    def test_load_strategy_component_module_imports_named_component(self) -> None:
+        definition = self.strategy_definitions["global_etf_rotation"]
+
+        module = load_strategy_component_module(
+            definition,
+            component_name="signal_logic",
+        )
+
+        self.assertEqual(module.__name__, "math")
+
+    def test_load_strategy_component_module_rejects_unknown_component(self) -> None:
+        definition = self.strategy_definitions["global_etf_rotation"]
+
+        with self.assertRaisesRegex(ValueError, "available components: signal_logic"):
+            load_strategy_component_module(
+                definition,
+                component_name="allocation",
+            )