feat: introduce multimodal trading feature pipeline

python/valuecell/agents/common/trading/_internal/coordinator.py

@@ -91,7 +91,7 @@ def __init__(
         self._request = request
         self.strategy_id = strategy_id
         self.portfolio_service = portfolio_service
-        self._features_pipeline = features_pipeline
+        self.features_pipeline = features_pipeline
         self._composer = composer
         self._execution_gateway = execution_gateway
         self._history_recorder = history_recorder
@@ -144,7 +144,7 @@ async def run_once(self) -> DecisionCycleResult:
             if self._request.exchange_config.market_type == MarketType.SPOT:
                 portfolio.buying_power = max(0.0, float(portfolio.account_balance))
 
-        pipeline_result = await self._features_pipeline.build()
+        pipeline_result = await self.features_pipeline.build()
         features = list(pipeline_result.features or [])
         market_features = extract_market_snapshot_features(features)
         digest = self._digest_builder.build(self._history_recorder.get_records())
@@ -612,7 +612,7 @@ async def close_all_positions(self) -> List[TradeHistoryEntry]:
             market_features: List[FeatureVector] = []
             if self._request.exchange_config.trading_mode == TradingMode.VIRTUAL:
                 try:
-                    pipeline_result = await self._features_pipeline.build()
+                    pipeline_result = await self.features_pipeline.build()
                     market_features = extract_market_snapshot_features(
                         pipeline_result.features or []
                     )

python/valuecell/agents/common/trading/base_agent.py

@@ -244,6 +244,15 @@ async def _run_background_decision(
 
         # Call user hook for custom initialization
         try:
+            # Initialize long-lived features resources if available
+            try:
+                await runtime.coordinator.features_pipeline.open()
+            except Exception:
+                logger.exception(
+                    "Error initializing features pipeline resources for strategy {}",
+                    strategy_id,
+                )
+
             await self._on_start(runtime, request)
         except Exception:
             logger.exception("Error in _on_start hook for strategy {}", strategy_id)
@@ -336,6 +345,13 @@ async def _run_background_decision(
                 )
 
             # Finalize: close resources and mark stopped/paused/error
+            try:
+                await runtime.coordinator.features_pipeline.close()
+            except Exception:
+                logger.exception(
+                    "Error closing features pipeline resources for strategy {}",
+                    strategy_id,
+                )
             await controller.finalize(
                 runtime, reason=stop_reason, reason_detail=stop_reason_detail
             )

python/valuecell/agents/common/trading/constants.py

@@ -15,3 +15,4 @@
 FEATURE_GROUP_BY_KEY = "group_by_key"
 FEATURE_GROUP_BY_INTERVAL_PREFIX = "interval_"
 FEATURE_GROUP_BY_MARKET_SNAPSHOT = "market_snapshot"
+FEATURE_GROUP_BY_IMAGE_ANALYSIS = "image_analysis"