Merge pull request #63 from AKKI0511/analyze-quanttradeai-for-new-feature-implementation

feat(streaming): add live trading orchestrator pipeline

Author: AKKI0511

README.md

@@ -140,7 +140,7 @@ Export the key and run the pipeline. A `sentiment_score` column is added when a
 
 ## Streaming (Optional)
 
-- CLI: `poetry run quanttradeai live-trade --url wss://... -c config/model_config.yaml`
+- CLI: `poetry run quanttradeai live-trade -m models/experiments/<run>/AAPL --config config/model_config.yaml --streaming-config config/streaming.yaml`
 - YAML‑driven gateway via `config/streaming.yaml`:
 
 ```yaml
@@ -167,6 +167,9 @@ gw.subscribe_to_trades(["AAPL"], lambda m: print("TRADE", m))
 - Provider adapters are discovered dynamically via `quanttradeai.streaming.providers.ProviderDiscovery`,
   validated with `ProviderConfigValidator`, and monitored through `ProviderHealthMonitor`. See
   [docs/api/streaming.md](docs/api/streaming.md) for detailed provider configuration and health tooling.
+- The live trading pipeline (`quanttradeai.streaming.live_trading.LiveTradingEngine`) combines the
+  streaming gateway, feature generation, model inference, risk controls, and optional health API. Use
+  `--health-api true` to expose `/health` and `/metrics` while streaming.
 
 ### Streaming Health Monitoring
 

docs/quick-reference.md

@@ -28,6 +28,14 @@ poetry run quanttradeai backtest-model -m models/experiments/<timestamp>/<SYMBOL
   -c config/model_config.yaml -b config/backtest_config.yaml \
   --cost-bps 5 --slippage-fixed 0.01 --liquidity-max-participation 0.25 \
   --skip-validation  # optional
+
+# Live trading (streaming)
+poetry run quanttradeai live-trade -m models/trained/<SYMBOL> \
+  -c config/model_config.yaml \
+  -s config/streaming.yaml \
+  --risk-config config/risk_config.yaml \
+  --position-manager-config config/position_manager.yaml \
+  --initial-capital 1000000 --min-history 220 --history-window 512
 ```
 
 ## 📊 Python API Patterns

poetry.lock

@@ -169,14 +169,65 @@ def cmd_backtest_model(
 
 @app.command("live-trade")
 def cmd_live_trade(
-    url: str = typer.Option(..., "--url", help="WebSocket URL for streaming data"),
-    config: str = typer.Option(
-        "config/model_config.yaml", "-c", "--config", help="Path to config file"
+    model_path: str = typer.Option(..., "-m", "--model-path", help="Saved model dir"),
+    model_config: str = typer.Option(
+        "config/model_config.yaml",
+        "-c",
+        "--config",
+        help="Path to model configuration file",
+    ),
+    streaming_config: str = typer.Option(
+        "config/streaming.yaml",
+        "-s",
+        "--streaming-config",
+        help="Path to streaming configuration file",
+    ),
+    risk_config: Optional[str] = typer.Option(
+        "config/risk_config.yaml",
+        "--risk-config",
+        help="Risk management configuration (drawdown/turnover)",
+    ),
+    position_manager_config: Optional[str] = typer.Option(
+        "config/position_manager.yaml",
+        "--position-manager-config",
+        help="Position manager configuration for live execution",
+    ),
+    health_api: Optional[bool] = typer.Option(
+        None,
+        "--health-api",
+        help="Override health API enable flag in streaming config (true/false)",
+    ),
+    initial_capital: float = typer.Option(
+        1_000_000.0, "--initial-capital", help="Starting capital for sizing"
+    ),
+    history_window: int = typer.Option(
+        512, "--history-window", help="History window for feature generation"
+    ),
+    min_history_for_features: int = typer.Option(
+        220,
+        "--min-history",
+        help="Minimum bars required before running inference",
+    ),
+    stop_loss_pct: float = typer.Option(
+        0.01, "--stop-loss-pct", help="Stop loss percentage applied to new trades"
     ),
 ):
     """Run real-time trading pipeline using streaming input."""
 
-    asyncio.run(run_live_pipeline(config, url))
+    asyncio.run(
+        run_live_pipeline(
+            model_config=model_config,
+            model_path=model_path,
+            streaming_config=streaming_config,
+            risk_config=risk_config,
+            position_manager_config=position_manager_config,
+            enable_health_api=health_api,
+            initial_capital=initial_capital,
+            history_window=history_window,
+            min_history_for_features=min_history_for_features,
+            stop_loss_pct=stop_loss_pct,
+        )
+    )
 
 
 @app.command("validate-config")

quanttradeai/cli.py

@@ -23,7 +23,6 @@
 
 from quanttradeai.data.loader import DataLoader
 from quanttradeai.data.processor import DataProcessor
-from quanttradeai.data.datasource import WebSocketDataSource
 from quanttradeai.backtest.backtester import simulate_trades, compute_metrics
 from quanttradeai.models.classifier import MomentumClassifier
 from quanttradeai.trading.drawdown_guard import DrawdownGuard
@@ -387,21 +386,36 @@ def evaluate_model(
         json.dump(results, f, indent=4)
 
 
-async def run_live_pipeline(config_path: str, url: str) -> None:
-    """Run a minimal real-time pipeline using WebSocket input.
-
-    Example
-    -------
-    >>> # asyncio.run(run_live_pipeline("config/model_config.yaml", "wss://example"))
-    """
-
-    processor = DataProcessor()
-    loader = DataLoader(config_path, data_source=WebSocketDataSource(url))
-
-    def handle(df: pd.DataFrame) -> None:
-        logger.info("Received update:\n%s", df.tail(1))
-
-    await loader.stream_data(processor, callback=handle)
+async def run_live_pipeline(
+    *,
+    model_config: str = "config/model_config.yaml",
+    model_path: str,
+    streaming_config: str = "config/streaming.yaml",
+    risk_config: str | None = "config/risk_config.yaml",
+    position_manager_config: str | None = "config/position_manager.yaml",
+    enable_health_api: bool | None = None,
+    initial_capital: float = 1_000_000.0,
+    history_window: int = 512,
+    min_history_for_features: int = 220,
+    stop_loss_pct: float = 0.01,
+) -> None:
+    """Run end-to-end real-time trading using the streaming gateway."""
+
+    from quanttradeai.streaming.live_trading import LiveTradingEngine
+
+    engine = LiveTradingEngine(
+        model_config=model_config,
+        model_path=model_path,
+        streaming_config=streaming_config,
+        risk_config=risk_config,
+        position_manager_config=position_manager_config,
+        enable_health_api=enable_health_api,
+        initial_capital=initial_capital,
+        history_window=history_window,
+        min_history_for_features=min_history_for_features,
+        stop_loss_pct=stop_loss_pct,
+    )
+    await engine.start()
 
 
 def _load_execution_cfg(

quanttradeai/main.py

@@ -12,6 +12,7 @@
     ProviderRegistry,
     StreamingProviderAdapter,
 )
+from .live_trading import LiveTradingEngine
 
 __all__ = [
     "StreamingGateway",
@@ -24,4 +25,5 @@
     "ProviderHealthMonitor",
     "ProviderRegistry",
     "StreamingProviderAdapter",
+    "LiveTradingEngine",
 ]