Upload 14 files

Dockerfile

@@ -1,18 +1,23 @@
-FROM python:3.10
+FROM python:3.11-slim
 
-RUN useradd -m -u 1000 user && python -m pip install --upgrade pip
-USER user
-ENV PATH="/home/user/.local/bin:$PATH"
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1
 
 WORKDIR /app
 
-COPY --chown=user ./requirements.txt requirements.txt
-RUN pip install --no-cache-dir --upgrade -r requirements.txt
+RUN useradd -m -u 1000 appuser
+
+COPY requirements.txt /app/requirements.txt
+RUN pip install --no-cache-dir -r /app/requirements.txt
+
+COPY backtrader /app/backtrader
+COPY app.py /app/app.py
 
-COPY --chown=user . /app
 ENV MCP_TRANSPORT=http
 ENV MCP_PORT=7860
 
 EXPOSE 7860
 
+USER appuser
+
 CMD ["python", "backtrader/mcp_output/start_mcp.py"]

README.md

@@ -1,10 +1,67 @@
 ---
-title: Backtrader
-emoji: 🦀
-colorFrom: green
-colorTo: gray
+title: backtrader MCP Service
+emoji: 🔧
+colorFrom: blue
+colorTo: indigo
 sdk: docker
 pinned: false
+license: mit
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# backtrader MCP Service
+
+This deployment package exposes selected local `backtrader` capabilities through MCP using FastMCP.
+
+## Available Tools
+
+- `health_check`
+- `list_modules`
+- `list_symbols`
+- `preview_csv_data`
+- `create_cerebro_session`
+- `run_sma_crossover_backtest`
+- `list_builtin_components`
+- `adapter_call_function`
+
+Detailed tool documentation is available in `backtrader/mcp_output/README_MCP.md`.
+
+## Local stdio usage
+
+1. Install dependencies:
+   ```bash
+   pip install -r requirements.txt
+   ```
+2. Start stdio MCP service:
+   ```bash
+   python backtrader/mcp_output/start_mcp.py
+   ```
+
+For direct local plugin entry:
+
+```bash
+python backtrader/mcp_output/mcp_plugin/main.py
+```
+
+## HTTP usage (Docker / HF Spaces style)
+
+Start with HTTP transport:
+
+```bash
+MCP_TRANSPORT=http MCP_PORT=7860 python backtrader/mcp_output/start_mcp.py
+```
+
+MCP endpoint:
+
+- `http://localhost:7860/mcp`
+
+## Docker usage
+
+```bash
+./run_docker.sh
+```
+
+or in PowerShell:
+
+```powershell
+./run_docker.ps1
+```

app.py

@@ -1,45 +1,59 @@
-from fastapi import FastAPI
+from __future__ import annotations
+
 import os
 import sys
+from pathlib import Path
+from typing import Any
+
+from fastapi import FastAPI
+
+BASE_DIR = Path(__file__).resolve().parent
+PLUGIN_DIR = BASE_DIR / "backtrader" / "mcp_output" / "mcp_plugin"
+
+if str(PLUGIN_DIR) not in sys.path:
+    sys.path.insert(0, str(PLUGIN_DIR))
+
+app = FastAPI(title="backtrader-mcp-info", version="1.0.0")
+
 
-mcp_plugin_path = os.path.join(os.path.dirname(__file__), "backtrader", "mcp_output", "mcp_plugin")
-sys.path.insert(0, mcp_plugin_path)
+def _extract_tool_info(tool_obj: Any) -> dict[str, str]:
+    name = getattr(tool_obj, "name", None) or getattr(tool_obj, "__name__", "unknown")
+    description = getattr(tool_obj, "description", None) or getattr(tool_obj, "__doc__", "") or ""
+    return {"name": str(name), "description": str(description).strip()}
 
-app = FastAPI(
-    title="Backtrader MCP Service",
-    description="Auto-generated MCP service for backtrader",
-    version="1.0.0"
-)
 
 @app.get("/")
-def root():
+def root() -> dict[str, Any]:
     return {
-        "service": "Backtrader MCP Service",
-        "version": "1.0.0",
-        "status": "running",
-        "transport": os.environ.get("MCP_TRANSPORT", "http")
+        "service": "backtrader MCP deployment",
+        "mcp_transport": os.getenv("MCP_TRANSPORT", "stdio"),
+        "mcp_port": int(os.getenv("MCP_PORT", "8000")),
+        "note": "This FastAPI app is supplementary and does not run the MCP server.",
     }
 
+
 @app.get("/health")
-def health_check():
-    return {"status": "healthy", "service": "backtrader MCP"}
+def health() -> dict[str, str]:
+    return {"status": "healthy"}
+
 
 @app.get("/tools")
-def list_tools():
-    try:
-        from mcp_service import create_app
-        mcp_app = create_app()
-        tools = []
-        for tool_name, tool_func in mcp_app.tools.items():
-            tools.append({
-                "name": tool_name,
-                "description": tool_func.__doc__ or "No description available"
-            })
-        return {"tools": tools}
-    except Exception as e:
-        return {"error": f"Failed to load tools: {str(e)}"}
+def tools() -> dict[str, Any]:
+    from mcp_service import create_app
+
+    mcp = create_app()
+    raw_tools = getattr(mcp, "tools", [])
+
+    if isinstance(raw_tools, dict):
+        items = [_extract_tool_info(value) for value in raw_tools.values()]
+    else:
+        items = [_extract_tool_info(item) for item in raw_tools]
+
+    return {"count": len(items), "tools": items}
+
 
 if __name__ == "__main__":
     import uvicorn
-    port = int(os.environ.get("PORT", 7860))
+
+    port = int(os.getenv("PORT", "7860"))
     uvicorn.run(app, host="0.0.0.0", port=port)

backtrader/mcp_output/README_MCP.md

@@ -1,147 +1,112 @@
-# Backtrader MCP (Model Context Protocol) Service README
-
-## 1) Project Introduction
-
-This MCP (Model Context Protocol) service exposes the core capabilities of the `backtrader` engine for strategy research and backtesting through callable service endpoints.
-
-Main functions:
-- Run backtests with `Cerebro`
-- Load market data (CSV, Pandas, Yahoo/Quandl-style feeds, and others)
-- Execute strategies, brokers, sizers, analyzers, and observers
-- Return structured performance outputs (returns, drawdown, Sharpe, trade stats, etc.)
-- Support optional plotting and integration-oriented workflows
-
-Repository analyzed: https://github.com/mementum/backtrader
-
----
-
-## 2) Installation Method
-
-### Requirements
-- Python runtime
-- Required: `matplotlib`
-- Common optional deps: `pandas`, `numpy`, `python-dateutil`, `pytz`
-- Integration-specific optional deps:
-  - IB stack (`ibpy`/`ib_insync` ecosystem) for IB-related store/broker/data
-  - Oanda client libs for Oanda store/broker/data
-  - `TA-Lib` for `backtrader.talib`
-  - `pyfolio` for PyFolio analyzer workflows
-
-### Install
-- Install backtrader:
-  pip install backtrader
-
-- Minimal plotting dependency:
-  pip install matplotlib
-
-- Recommended data stack:
-  pip install pandas numpy python-dateutil pytz
-
----
-
-## 3) Quick Start
-
-### Minimal service workflow
-1. Create a backtest run request
-2. Provide data source (e.g., CSV or Pandas)
-3. Select strategy and parameters
-4. Attach analyzers (Sharpe, DrawDown, Returns, TradeAnalyzer, etc.)
-5. Execute and fetch normalized results
-
-### Typical Python usage pattern behind the service
-import backtrader as bt
-
-class SmaCross(bt.Strategy):
-    params = dict(fast=10, slow=30)
-    def __init__(self):
-        sma1 = bt.ind.SMA(period=self.p.fast)
-        sma2 = bt.ind.SMA(period=self.p.slow)
-        self.crossover = bt.ind.CrossOver(sma1, sma2)
-    def next(self):
-        if not self.position and self.crossover > 0:
-            self.buy()
-        elif self.position and self.crossover < 0:
-            self.close()
-
-cerebro = bt.Cerebro()
-cerebro.addstrategy(SmaCross)
-data = bt.feeds.GenericCSVData(dataname="datas/2006-day-001.txt")
-cerebro.adddata(data)
-cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name="sharpe")
-cerebro.run()
-
-In the MCP (Model Context Protocol) service, this is wrapped into endpoint calls (see below).
-
----
-
-## 4) Available Tools and Endpoints List
-
-Recommended service endpoints (developer-oriented mapping to backtrader modules):
-
-- `health`
-  - Basic readiness/liveness check.
-
-- `list_capabilities`
-  - Returns supported feeds, analyzers, observers, sizers, and integration availability.
-
-- `run_backtest`
-  - Core endpoint. Runs a backtest with:
-    - strategy class/name + params
-    - data config (feed type, path, timeframe/compression)
-    - broker/cash/commission/slippage
-    - analyzers/observers/sizers
-  - Returns metrics, trades, and optional artifacts metadata.
-
-- `optimize_strategy`
-  - Runs parameter sweeps/optimization over strategy params.
-  - Returns ranked result set and top configurations.
-
-- `load_data_preview`
-  - Validates data config and previews parsed bars/date range before execution.
-
-- `list_analyzers`
-  - Lists available analyzers (e.g., `SharpeRatio`, `DrawDown`, `Returns`, `TradeAnalyzer`, `TimeReturn`, `SQN`, `VWR`).
-
-- `list_feeds`
-  - Lists supported feed adapters (CSV generic/backtrader CSV, Pandas, Yahoo/Quandl-style, IB/Oanda/VC when enabled).
-
-- `run_btrun_cli` (optional bridge)
-  - Wraps `backtrader.btrun.btrun` style execution for CLI-compatible scenarios.
-
----
-
-## 5) Common Issues and Notes
-
-- Dependency gaps:
-  - Core backtesting works with minimal deps, but many feeds/integrations need extra packages.
-- TA-Lib and PyFolio:
-  - Optional; endpoint should gracefully report “not installed” capability.
-- Broker/store integrations:
-  - IB/Oanda/VC modules require external vendor/client ecosystems and credentials.
-- Data alignment/timeframes:
-  - Multi-timeframe and replay/resample logic can be sensitive to feed configuration.
-- Plotting in server environments:
-  - Prefer non-interactive backends or disable plotting in headless deployments.
-- Performance:
-  - Strategy optimization can be CPU-heavy; use bounded parameter grids and job limits.
-- Stability:
-  - Import feasibility is high and intrusiveness risk is low per analysis, but integration modules vary by environment.
-
----
-
-## 6) Reference Links / Documentation
-
-- Upstream repository: https://github.com/mementum/backtrader
-- Package entry points of interest:
-  - `backtrader.cerebro.Cerebro`
-  - `backtrader.strategy.Strategy`
-  - `backtrader.feed` / `backtrader.feeds.*`
-  - `backtrader.analyzers.*`
-  - `backtrader.observers.*`
-  - `backtrader.btrun.btrun`
-- Samples directory (practical patterns):
-  - `samples/` in repository root
-- Tests directory (behavior references):
-  - `tests/` in repository root
-
-If you want, I can also generate a concrete endpoint I/O schema (JSON request/response shapes) for each MCP (Model Context Protocol) service endpoint.
+# Backtrader MCP Plugin
+
+This MCP plugin exposes core local backtesting capabilities from the `backtrader` library.
+
+## Exposed Tools
+
+### 1) `health_check`
+- **Description**: Check service and dependency availability.
+- **Parameters**: none
+- **Example**:
+```json
+{"tool":"health_check","arguments":{}}
+```
+
+### 2) `list_modules`
+- **Description**: List loaded and failed `backtrader` modules tracked by the adapter.
+- **Parameters**: none
+- **Example**:
+```json
+{"tool":"list_modules","arguments":{}}
+```
+
+### 3) `list_symbols`
+- **Description**: Inspect symbols exported by a loaded module.
+- **Parameters**:
+  - `module_name` (string): module name, e.g. `backtrader.indicators`
+  - `include_private` (boolean, optional)
+  - `limit` (integer, optional)
+- **Example**:
+```json
+{"tool":"list_symbols","arguments":{"module_name":"backtrader.indicators","limit":30}}
+```
+
+### 4) `preview_csv_data`
+- **Description**: Read a quick sample of local CSV rows.
+- **Parameters**:
+  - `data_path` (string): absolute or relative CSV path
+  - `delimiter` (string, optional)
+  - `max_rows` (integer, optional)
+- **Example**:
+```json
+{"tool":"preview_csv_data","arguments":{"data_path":"2006-day-001.txt","delimiter":",","max_rows":5}}
+```
+
+### 5) `create_cerebro_session`
+- **Description**: Create and summarize a baseline `Cerebro` session.
+- **Parameters**:
+  - `initial_cash` (number, optional)
+  - `commission` (number, optional)
+- **Example**:
+```json
+{"tool":"create_cerebro_session","arguments":{"initial_cash":10000,"commission":0.001}}
+```
+
+### 6) `run_sma_crossover_backtest`
+- **Description**: Run a simple SMA crossover strategy on local CSV data.
+- **Parameters**:
+  - `data_path` (string)
+  - `fast_period` (integer, optional)
+  - `slow_period` (integer, optional)
+  - `initial_cash` (number, optional)
+  - `commission` (number, optional)
+  - `stake` (integer, optional)
+- **Example**:
+```json
+{"tool":"run_sma_crossover_backtest","arguments":{"data_path":"2006-day-001.txt","fast_period":10,"slow_period":30}}
+```
+
+### 7) `list_builtin_components`
+- **Description**: List built-in indicators, analyzers, or sizers.
+- **Parameters**:
+  - `component_type` (string): `indicators` | `analyzers` | `sizers`
+  - `limit` (integer, optional)
+- **Example**:
+```json
+{"tool":"list_builtin_components","arguments":{"component_type":"indicators","limit":25}}
+```
+
+### 8) `adapter_call_function`
+- **Description**: Generic adapter function call for advanced introspection.
+- **Parameters**:
+  - `module_name` (string)
+  - `function_name` (string)
+  - `positional_args_json` (string, optional JSON array)
+  - `keyword_args_json` (string, optional JSON object)
+- **Example**:
+```json
+{"tool":"adapter_call_function","arguments":{"module_name":"backtrader.mathsupport","function_name":"average","positional_args_json":"[[1,2,3]]"}}
+```
+
+## Running Locally (stdio)
+
+```bash
+cd backtrader/mcp_output
+python start_mcp.py
+```
+
+Or run the explicit stdio local entry point:
+
+```bash
+cd backtrader/mcp_output/mcp_plugin
+python main.py
+```
+
+## Running via HTTP
+
+```bash
+cd backtrader/mcp_output
+MCP_TRANSPORT=http MCP_PORT=7860 python start_mcp.py
+```
+
+The MCP HTTP endpoint is served at `/mcp`.

backtrader/mcp_output/mcp_plugin/adapter.py

@@ -1,472 +1,193 @@
-import os
-import sys
-import importlib
-import importlib.util
-import importlib.machinery
-from typing import Any, Dict, Optional, Tuple
-
-source_path = os.path.join(
-    os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))),
-    "source",
-)
-sys.path.insert(0, source_path)
+from __future__ import annotations
 
+import importlib
+import inspect
+import pkgutil
+import sys
+from pathlib import Path
+from types import ModuleType
+from typing import Any
 
-class Adapter:
-    """
-    MCP Import Mode Adapter for backtrader repository integration.
-
-    This adapter attempts direct module import first ("import" mode) and falls back
-    to a file-based dynamic loader ("fallback_cli") when needed.
+_PLUGIN_DIR = Path(__file__).resolve().parent
+_REPO_DIR = _PLUGIN_DIR.parent.parent
+_SOURCE_DIR = _REPO_DIR / "source"
 
-    Unified return format for all public methods:
-    {
-        "status": "success" | "error",
-        "mode": "<current mode>",
-        "message": "<human-readable summary>",
-        "data": <optional payload>,
-        "error": "<optional error details>"
-    }
-    """
+if str(_SOURCE_DIR) not in sys.path:
+    sys.path.insert(0, str(_SOURCE_DIR))
 
-    def __init__(self) -> None:
-        self.mode = "import"
-        self._loaded_modules: Dict[str, Any] = {}
-        self._import_errors: Dict[str, str] = {}
-        self._initialize_imports()
 
-    # -------------------------------------------------------------------------
-    # Core helpers
-    # -------------------------------------------------------------------------
-    def _ok(self, message: str, data: Optional[Any] = None) -> Dict[str, Any]:
+class Adapter:
+    def __init__(self, package_name: str = "backtrader") -> None:
+        self.package_name = package_name
+        self._root_module: ModuleType | None = None
+        self._loaded_modules: dict[str, ModuleType] = {}
+        self._failed_modules: dict[str, str] = {}
+        self._mode = "normal"
+        self._bootstrap()
+
+    def _bootstrap(self) -> None:
+        try:
+            self._root_module = importlib.import_module(self.package_name)
+            self._loaded_modules[self.package_name] = self._root_module
+        except Exception as exc:
+            self._mode = "blackbox"
+            self._failed_modules[self.package_name] = str(exc)
+            return
+
+        self._import_submodules()
+        if len(self._loaded_modules) <= 0:
+            self._mode = "blackbox"
+
+    def _import_submodules(self) -> None:
+        if self._root_module is None:
+            return
+
+        module_paths = getattr(self._root_module, "__path__", None)
+        if not module_paths:
+            return
+
+        for module_info in pkgutil.walk_packages(module_paths, prefix=f"{self.package_name}."):
+            module_name = module_info.name
+            if module_name in self._loaded_modules:
+                continue
+
+            try:
+                module = importlib.import_module(module_name)
+                self._loaded_modules[module_name] = module
+            except Exception as exc:
+                self._failed_modules[module_name] = str(exc)
+
+        if not self._loaded_modules:
+            self._mode = "blackbox"
+
+    def health(self) -> dict[str, Any]:
+        status = "ok" if self._loaded_modules else "fallback"
         return {
-            "status": "success",
-            "mode": self.mode,
-            "message": message,
-            "data": data,
+            "status": status,
+            "mode": self._mode,
+            "package": self.package_name,
+            "source_dir": str(_SOURCE_DIR),
+            "loaded_count": len(self._loaded_modules),
+            "failed_count": len(self._failed_modules),
+            "loaded_modules": sorted(self._loaded_modules.keys()),
+            "failed_modules": self._failed_modules,
         }
 
-    def _err(self, message: str, error: Optional[Exception] = None) -> Dict[str, Any]:
+    def list_modules(self) -> dict[str, Any]:
+        status = "ok" if self._loaded_modules else "fallback"
         return {
-            "status": "error",
-            "mode": self.mode,
-            "message": message,
-            "error": str(error) if error else None,
+            "status": status,
+            "mode": self._mode,
+            "loaded": sorted(self._loaded_modules.keys()),
+            "failed": self._failed_modules,
         }
 
-    def _safe_import(self, module_path: str) -> Tuple[Optional[Any], Optional[str]]:
-        try:
-            module = importlib.import_module(module_path)
-            self._loaded_modules[module_path] = module
-            return module, None
-        except Exception as exc:
-            self._import_errors[module_path] = str(exc)
-            return None, str(exc)
-
-    def _load_module_from_file(self, module_name: str, file_path: str) -> Tuple[Optional[Any], Optional[str]]:
-        try:
-            if not os.path.exists(file_path):
-                return None, f"File not found: {file_path}. Verify repository extraction path and source directory."
-            loader = importlib.machinery.SourceFileLoader(module_name, file_path)
-            spec = importlib.util.spec_from_loader(module_name, loader)
-            if spec is None:
-                return None, f"Unable to create module spec for {module_name}."
-            mod = importlib.util.module_from_spec(spec)
-            loader.exec_module(mod)
-            self._loaded_modules[module_name] = mod
-            return mod, None
-        except Exception as exc:
-            return None, str(exc)
-
-    def _initialize_imports(self) -> None:
-        targets = [
-            "backtrader.btrun.btrun",
-            "tools.rewrite-data",
-            "tools.yahoodownload",
-            "contrib.utils.iqfeed-to-influxdb",
-            "contrib.utils.influxdb-import",
-            "contrib.samples.pair-trading.pair-trading",
-            "samples.weekdays-filler.weekdaysfiller",
-            "samples.weekdays-filler.weekdaysaligner",
-        ]
-
-        for mod in targets:
-            self._safe_import(mod)
-
-        if any(m not in self._loaded_modules for m in targets):
-            self.mode = "fallback_cli"
-
-    def get_health(self) -> Dict[str, Any]:
-        """
-        Return adapter health and module import diagnostics.
-
-        Returns:
-            dict: Unified status payload with import success/failure overview.
-        """
-        return self._ok(
-            "Adapter health check completed.",
-            data={
-                "loaded_modules": list(self._loaded_modules.keys()),
-                "import_errors": self._import_errors,
-                "source_path": source_path,
-            },
-        )
-
-    # -------------------------------------------------------------------------
-    # backtrader.btrun.btrun
-    # -------------------------------------------------------------------------
-    def call_btrun(self, argv: Optional[list] = None) -> Dict[str, Any]:
-        """
-        Execute the built-in btrun command module.
-
-        Args:
-            argv (list, optional): Command-line style arguments. If omitted, module defaults are used.
-
-        Returns:
-            dict: Unified status with execution result or actionable error guidance.
-        """
-        try:
-            module = self._loaded_modules.get("backtrader.btrun.btrun")
-            if module is None:
-                module, err = self._safe_import("backtrader.btrun.btrun")
-                if module is None:
-                    return self._err(
-                        "Failed to import backtrader btrun module. Confirm source/backtrader is present and importable.",
-                        Exception(err),
-                    )
-
-            if hasattr(module, "main"):
-                result = module.main(argv) if argv is not None else module.main()
-                return self._ok("btrun executed via main().", data={"result": result})
-
-            return self._err(
-                "btrun module does not expose main(). Use backtrader.btrun.btrun manually with repository-compatible arguments."
-            )
-        except Exception as exc:
-            return self._err("btrun execution failed. Validate arguments and data file paths.", exc)
-
-    # -------------------------------------------------------------------------
-    # tools/rewrite-data.py
-    # -------------------------------------------------------------------------
-    def call_rewrite_data_parse_args(self, args: Optional[list] = None) -> Dict[str, Any]:
-        """
-        Call parse_args from tools/rewrite-data.py.
-
-        Args:
-            args (list, optional): Argument vector for parser input.
-
-        Returns:
-            dict: Unified status with parser namespace/details.
-        """
-        try:
-            mod = self._loaded_modules.get("tools.rewrite-data")
-            if mod is None:
-                path = os.path.join(source_path, "tools", "rewrite-data.py")
-                mod, err = self._load_module_from_file("tools.rewrite_data_dynamic", path)
-                if mod is None:
-                    return self._err("Unable to load tools/rewrite-data.py for parse_args.", Exception(err))
-
-            fn = getattr(mod, "parse_args", None)
-            if fn is None:
-                return self._err("parse_args not found in tools/rewrite-data.py. Confirm repository version compatibility.")
-            res = fn(args) if args is not None else fn()
-            return self._ok("rewrite-data parse_args executed.", data={"result": res})
-        except Exception as exc:
-            return self._err("rewrite-data parse_args failed. Check argument format.", exc)
-
-    def call_rewrite_data_runstrat(self, *args: Any, **kwargs: Any) -> Dict[str, Any]:
-        """
-        Call runstrat from tools/rewrite-data.py.
-
-        Args:
-            *args: Positional arguments forwarded to runstrat.
-            **kwargs: Keyword arguments forwarded to runstrat.
-
-        Returns:
-            dict: Unified status with execution output.
-        """
-        try:
-            mod = self._loaded_modules.get("tools.rewrite-data")
-            if mod is None:
-                path = os.path.join(source_path, "tools", "rewrite-data.py")
-                mod, err = self._load_module_from_file("tools.rewrite_data_dynamic", path)
-                if mod is None:
-                    return self._err("Unable to load tools/rewrite-data.py for runstrat.", Exception(err))
-
-            fn = getattr(mod, "runstrat", None)
-            if fn is None:
-                return self._err("runstrat not found in tools/rewrite-data.py. Confirm repository version compatibility.")
-            res = fn(*args, **kwargs)
-            return self._ok("rewrite-data runstrat executed.", data={"result": res})
-        except Exception as exc:
-            return self._err("rewrite-data runstrat failed. Validate strategy/data parameters.", exc)
-
-    # -------------------------------------------------------------------------
-    # tools/yahoodownload.py
-    # -------------------------------------------------------------------------
-    def create_yahoo_download_instance(self, *args: Any, **kwargs: Any) -> Dict[str, Any]:
-        """
-        Instantiate YahooDownload from tools/yahoodownload.py.
-
-        Args:
-            *args: Constructor positional arguments.
-            **kwargs: Constructor keyword arguments.
-
-        Returns:
-            dict: Unified status with instantiated object.
-        """
-        try:
-            mod = self._loaded_modules.get("tools.yahoodownload")
-            if mod is None:
-                mod, err = self._safe_import("tools.yahoodownload")
-                if mod is None:
-                    return self._err("Unable to import tools.yahoodownload.", Exception(err))
-
-            cls = getattr(mod, "YahooDownload", None)
-            if cls is None:
-                return self._err("YahooDownload class not found in tools.yahoodownload.")
-            inst = cls(*args, **kwargs)
-            return self._ok("YahooDownload instance created.", data={"instance": inst})
-        except Exception as exc:
-            return self._err("Failed to create YahooDownload instance. Check constructor parameters.", exc)
-
-    def call_yahoodownload_parse_args(self, args: Optional[list] = None) -> Dict[str, Any]:
-        """
-        Call parse_args from tools/yahoodownload.py.
-
-        Args:
-            args (list, optional): Argument vector for parser input.
-
-        Returns:
-            dict: Unified status with parser output.
-        """
-        try:
-            mod = self._loaded_modules.get("tools.yahoodownload")
-            if mod is None:
-                mod, err = self._safe_import("tools.yahoodownload")
-                if mod is None:
-                    return self._err("Unable to import tools.yahoodownload.", Exception(err))
-
-            fn = getattr(mod, "parse_args", None)
-            if fn is None:
-                return self._err("parse_args not found in tools.yahoodownload.")
-            res = fn(args) if args is not None else fn()
-            return self._ok("yahoodownload parse_args executed.", data={"result": res})
-        except Exception as exc:
-            return self._err("yahoodownload parse_args failed. Verify CLI-style arguments.", exc)
-
-    # -------------------------------------------------------------------------
-    # contrib/utils/iqfeed-to-influxdb.py
-    # -------------------------------------------------------------------------
-    def create_iqfeed_tool_instance(self, *args: Any, **kwargs: Any) -> Dict[str, Any]:
-        """
-        Instantiate IQFeedTool from contrib/utils/iqfeed-to-influxdb.py.
-
-        Args:
-            *args: Constructor positional arguments.
-            **kwargs: Constructor keyword arguments.
-
-        Returns:
-            dict: Unified status with created instance.
-        """
-        try:
-            mod = self._loaded_modules.get("contrib.utils.iqfeed-to-influxdb")
-            if mod is None:
-                path = os.path.join(source_path, "contrib", "utils", "iqfeed-to-influxdb.py")
-                mod, err = self._load_module_from_file("contrib.utils.iqfeed_to_influxdb_dynamic", path)
-                if mod is None:
-                    return self._err("Unable to load contrib/utils/iqfeed-to-influxdb.py.", Exception(err))
-
-            cls = getattr(mod, "IQFeedTool", None)
-            if cls is None:
-                return self._err("IQFeedTool class not found in contrib/utils/iqfeed-to-influxdb.py.")
-            inst = cls(*args, **kwargs)
-            return self._ok("IQFeedTool instance created.", data={"instance": inst})
-        except Exception as exc:
-            return self._err("Failed to create IQFeedTool instance. Validate required external service settings.", exc)
+    def list_symbols(self, module_name: str, include_private: bool = False, limit: int = 200) -> dict[str, Any]:
+        module = self._loaded_modules.get(module_name)
+        if module is None:
+            error_message = f"Module not loaded: {module_name}"
+            if module_name in self._failed_modules:
+                error_message = f"Module failed to import: {module_name} ({self._failed_modules[module_name]})"
+            return {"status": "error", "error": error_message}
+
+        names = dir(module)
+        if not include_private:
+            names = [name for name in names if not name.startswith("_")]
+
+        symbols: list[dict[str, str]] = []
+        for name in names[: max(limit, 0)]:
+            try:
+                value = getattr(module, name)
+                if inspect.isclass(value):
+                    kind = "class"
+                elif inspect.isfunction(value) or inspect.ismethod(value):
+                    kind = "function"
+                elif inspect.ismodule(value):
+                    kind = "module"
+                else:
+                    kind = "value"
+                symbols.append({"name": name, "type": kind})
+            except Exception:
+                symbols.append({"name": name, "type": "unknown"})
 
-    # -------------------------------------------------------------------------
-    # contrib/utils/influxdb-import.py
-    # -------------------------------------------------------------------------
-    def create_influxdb_tool_instance(self, *args: Any, **kwargs: Any) -> Dict[str, Any]:
-        """
-        Instantiate InfluxDBTool from contrib/utils/influxdb-import.py.
-
-        Args:
-            *args: Constructor positional arguments.
-            **kwargs: Constructor keyword arguments.
-
-        Returns:
-            dict: Unified status with created instance.
-        """
-        try:
-            mod = self._loaded_modules.get("contrib.utils.influxdb-import")
-            if mod is None:
-                path = os.path.join(source_path, "contrib", "utils", "influxdb-import.py")
-                mod, err = self._load_module_from_file("contrib.utils.influxdb_import_dynamic", path)
-                if mod is None:
-                    return self._err("Unable to load contrib/utils/influxdb-import.py.", Exception(err))
-
-            cls = getattr(mod, "InfluxDBTool", None)
-            if cls is None:
-                return self._err("InfluxDBTool class not found in contrib/utils/influxdb-import.py.")
-            inst = cls(*args, **kwargs)
-            return self._ok("InfluxDBTool instance created.", data={"instance": inst})
-        except Exception as exc:
-            return self._err("Failed to create InfluxDBTool instance. Verify InfluxDB connection and credentials.", exc)
-
-    # -------------------------------------------------------------------------
-    # contrib/samples/pair-trading/pair-trading.py
-    # -------------------------------------------------------------------------
-    def create_pair_trading_strategy_instance(self, *args: Any, **kwargs: Any) -> Dict[str, Any]:
-        """
-        Instantiate PairTradingStrategy from contrib/samples/pair-trading/pair-trading.py.
-
-        Args:
-            *args: Constructor positional arguments.
-            **kwargs: Constructor keyword arguments.
-
-        Returns:
-            dict: Unified status with created strategy instance.
-        """
-        try:
-            mod = self._loaded_modules.get("contrib.samples.pair-trading.pair-trading")
-            if mod is None:
-                path = os.path.join(source_path, "contrib", "samples", "pair-trading", "pair-trading.py")
-                mod, err = self._load_module_from_file("contrib.samples.pair_trading_dynamic", path)
-                if mod is None:
-                    return self._err("Unable to load pair-trading sample module.", Exception(err))
-
-            cls = getattr(mod, "PairTradingStrategy", None)
-            if cls is None:
-                return self._err("PairTradingStrategy class not found in pair-trading sample module.")
-            inst = cls(*args, **kwargs)
-            return self._ok("PairTradingStrategy instance created.", data={"instance": inst})
-        except Exception as exc:
-            return self._err("Failed to create PairTradingStrategy instance. Use Cerebro-managed instantiation for runtime use.", exc)
-
-    def call_pair_trading_parse_args(self, args: Optional[list] = None) -> Dict[str, Any]:
-        """
-        Call parse_args from pair-trading sample module.
-
-        Args:
-            args (list, optional): Parser argument list.
-
-        Returns:
-            dict: Unified status with parser result.
-        """
-        try:
-            path = os.path.join(source_path, "contrib", "samples", "pair-trading", "pair-trading.py")
-            mod, err = self._load_module_from_file("contrib.samples.pair_trading_dynamic", path)
-            if mod is None:
-                return self._err("Unable to load pair-trading sample for parse_args.", Exception(err))
-
-            fn = getattr(mod, "parse_args", None)
-            if fn is None:
-                return self._err("parse_args not found in pair-trading sample module.")
-            res = fn(args) if args is not None else fn()
-            return self._ok("pair-trading parse_args executed.", data={"result": res})
-        except Exception as exc:
-            return self._err("pair-trading parse_args failed. Verify provided options.", exc)
-
-    def call_pair_trading_runstrategy(self, *args: Any, **kwargs: Any) -> Dict[str, Any]:
-        """
-        Call runstrategy from pair-trading sample module.
-
-        Args:
-            *args: Positional arguments for runstrategy.
-            **kwargs: Keyword arguments for runstrategy.
-
-        Returns:
-            dict: Unified status with run output.
-        """
-        try:
-            path = os.path.join(source_path, "contrib", "samples", "pair-trading", "pair-trading.py")
-            mod, err = self._load_module_from_file("contrib.samples.pair_trading_dynamic", path)
-            if mod is None:
-                return self._err("Unable to load pair-trading sample for runstrategy.", Exception(err))
-
-            fn = getattr(mod, "runstrategy", None)
-            if fn is None:
-                return self._err("runstrategy not found in pair-trading sample module.")
-            res = fn(*args, **kwargs)
-            return self._ok("pair-trading runstrategy executed.", data={"result": res})
-        except Exception as exc:
-            return self._err("pair-trading runstrategy failed. Check data feeds and broker settings.", exc)
-
-    # -------------------------------------------------------------------------
-    # samples/weekdays-filler
-    # -------------------------------------------------------------------------
-    def create_weekdays_filler_instance(self, *args: Any, **kwargs: Any) -> Dict[str, Any]:
-        """
-        Instantiate WeekDaysFiller from samples/weekdays-filler/weekdaysfiller.py.
+        return {
+            "status": "ok",
+            "module": module_name,
+            "count": len(symbols),
+            "symbols": symbols,
+        }
 
-        Args:
-            *args: Constructor positional arguments.
-            **kwargs: Constructor keyword arguments.
+    def call_function(
+        self,
+        module_name: str,
+        function_name: str,
+        positional_args: list[Any] | None = None,
+        keyword_args: dict[str, Any] | None = None,
+    ) -> dict[str, Any]:
+        module = self._loaded_modules.get(module_name)
+        if module is None:
+            return {"status": "error", "error": f"Module not loaded: {module_name}"}
 
-        Returns:
-            dict: Unified status with created instance.
-        """
-        try:
-            path = os.path.join(source_path, "samples", "weekdays-filler", "weekdaysfiller.py")
-            mod, err = self._load_module_from_file("samples.weekdays_filler_dynamic", path)
-            if mod is None:
-                return self._err("Unable to load weekdaysfiller sample module.", Exception(err))
+        if not hasattr(module, function_name):
+            return {"status": "error", "error": f"Function not found: {module_name}.{function_name}"}
 
-            cls = getattr(mod, "WeekDaysFiller", None)
-            if cls is None:
-                return self._err("WeekDaysFiller class not found in weekdaysfiller sample module.")
-            inst = cls(*args, **kwargs)
-            return self._ok("WeekDaysFiller instance created.", data={"instance": inst})
-        except Exception as exc:
-            return self._err("Failed to create WeekDaysFiller instance. Validate constructor inputs.", exc)
+        target = getattr(module, function_name)
+        if not callable(target):
+            return {"status": "error", "error": f"Target is not callable: {module_name}.{function_name}"}
 
-    def call_weekdaysaligner_parse_args(self, args: Optional[list] = None) -> Dict[str, Any]:
-        """
-        Call parse_args from samples/weekdays-filler/weekdaysaligner.py.
+        args = positional_args if positional_args is not None else []
+        kwargs = keyword_args if keyword_args is not None else {}
 
-        Args:
-            args (list, optional): Parser argument list.
-
-        Returns:
-            dict: Unified status with parser namespace/output.
-        """
         try:
-            path = os.path.join(source_path, "samples", "weekdays-filler", "weekdaysaligner.py")
-            mod, err = self._load_module_from_file("samples.weekdays_aligner_dynamic", path)
-            if mod is None:
-                return self._err("Unable to load weekdaysaligner sample module.", Exception(err))
-
-            fn = getattr(mod, "parse_args", None)
-            if fn is None:
-                return self._err("parse_args not found in weekdaysaligner sample module.")
-            res = fn(args) if args is not None else fn()
-            return self._ok("weekdaysaligner parse_args executed.", data={"result": res})
+            result = target(*args, **kwargs)
+            return {
+                "status": "ok",
+                "module": module_name,
+                "function": function_name,
+                "result": result,
+            }
         except Exception as exc:
-            return self._err("weekdaysaligner parse_args failed. Review argument values.", exc)
-
-    def call_weekdaysaligner_runstrat(self, *args: Any, **kwargs: Any) -> Dict[str, Any]:
-        """
-        Call runstrat from samples/weekdays-filler/weekdaysaligner.py.
+            return {
+                "status": "error",
+                "module": module_name,
+                "function": function_name,
+                "error": str(exc),
+            }
+
+    def create_instance(
+        self,
+        module_name: str,
+        class_name: str,
+        positional_args: list[Any] | None = None,
+        keyword_args: dict[str, Any] | None = None,
+    ) -> dict[str, Any]:
+        module = self._loaded_modules.get(module_name)
+        if module is None:
+            return {"status": "error", "error": f"Module not loaded: {module_name}"}
+
+        if not hasattr(module, class_name):
+            return {"status": "error", "error": f"Class not found: {module_name}.{class_name}"}
+
+        target = getattr(module, class_name)
+        if not inspect.isclass(target):
+            return {"status": "error", "error": f"Target is not a class: {module_name}.{class_name}"}
+
+        args = positional_args if positional_args is not None else []
+        kwargs = keyword_args if keyword_args is not None else {}
 
-        Args:
-            *args: Positional args forwarded to runstrat.
-            **kwargs: Keyword args forwarded to runstrat.
-
-        Returns:
-            dict: Unified status with strategy run result.
-        """
         try:
-            path = os.path.join(source_path, "samples", "weekdays-filler", "weekdaysaligner.py")
-            mod, err = self._load_module_from_file("samples.weekdays_aligner_dynamic", path)
-            if mod is None:
-                return self._err("Unable to load weekdaysaligner sample module.", Exception(err))
-
-            fn = getattr(mod, "runstrat", None)
-            if fn is None:
-                return self._err("runstrat not found in weekdaysaligner sample module.")
-            res = fn(*args, **kwargs)
-            return self._ok("weekdaysaligner runstrat executed.", data={"result": res})
+            instance = target(*args, **kwargs)
+            return {
+                "status": "ok",
+                "module": module_name,
+                "class": class_name,
+                "instance_type": type(instance).__name__,
+                "instance_repr": repr(instance),
+            }
         except Exception as exc:
-            return self._err("weekdaysaligner runstrat failed. Verify feed and calendar options.", exc)
+            return {
+                "status": "error",
+                "module": module_name,
+                "class": class_name,
+                "error": str(exc),
+            }

backtrader/mcp_output/mcp_plugin/main.py

@@ -1,13 +1,9 @@
-"""
-MCP Service Auto-Wrapper - Auto-generated
-"""
+from __future__ import annotations
+
+# Local stdio entry point only (Claude Desktop / CLI); not for web or Docker deployment.
 from mcp_service import create_app
 
-def main():
-    """Main entry point"""
-    app = create_app()
-    return app
 
 if __name__ == "__main__":
-    app = main()
-    app.run()
+    app = create_app()
+    app.run(transport="stdio")

backtrader/mcp_output/mcp_plugin/mcp_service.py

@@ -1,303 +1,347 @@
-import os
+from __future__ import annotations
+
+import csv
+import json
 import sys
+from pathlib import Path
+from typing import Any
 
-source_path = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), "source")
-if source_path not in sys.path:
-    sys.path.insert(0, source_path)
+_PLUGIN_DIR = Path(__file__).resolve().parent
+_REPO_DIR = _PLUGIN_DIR.parent.parent
+_SOURCE_DIR = _REPO_DIR / "source"
 
-from fastmcp import FastMCP
+if str(_SOURCE_DIR) not in sys.path:
+    sys.path.insert(0, str(_SOURCE_DIR))
 
-from tools.rewrite-data import parse_args, runstrat
-from tools.yahoodownload import parse_args, YahooDownload
-from contrib.utils.iqfeed-to-influxdb import IQFeedTool
-from contrib.utils.influxdb-import import InfluxDBTool
-from contrib.samples.pair-trading import runstrategy, parse_args, PairTradingStrategy
-from samples.weekdays-filler.weekdaysfiller import WeekDaysFiller
-from samples.weekdays-filler.weekdaysaligner import parse_args, runstrat
+try:
+    from fastmcp import FastMCP
+except Exception:
+    FastMCP = None
 
-mcp = FastMCP("unknown_service")
+try:
+    import backtrader as bt
+except Exception:
+    bt = None
 
+try:
+    import backtrader.feeds as btfeeds
+except Exception:
+    btfeeds = None
 
-@mcp.tool(name="parse_args", description="Auto-wrapped function parse_args")
-def parse_args(payload: dict):
-    try:
-        if parse_args is None:
-            return {"success": False, "result": None, "error": "Function parse_args is not available"}
-        result = parse_args(**payload)
-        return {"success": True, "result": result, "error": None}
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-@mcp.tool(name="runstrat", description="Auto-wrapped function runstrat")
-def runstrat(payload: dict):
-    try:
-        if runstrat is None:
-            return {"success": False, "result": None, "error": "Function runstrat is not available"}
-        result = runstrat(**payload)
-        return {"success": True, "result": result, "error": None}
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-@mcp.tool(name="parse_args", description="Auto-wrapped function parse_args")
-def parse_args(payload: dict):
-    try:
-        if parse_args is None:
-            return {"success": False, "result": None, "error": "Function parse_args is not available"}
-        result = parse_args(**payload)
-        return {"success": True, "result": result, "error": None}
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-@mcp.tool(name="yahoodownload", description="YahooDownload class")
-def yahoodownload(*args, **kwargs):
-    """YahooDownload class"""
-    try:
-        if YahooDownload is None:
-            return {"success": False, "result": None, "error": "Class YahooDownload is not available, path may need adjustment"}
-        
-        # MCP parameter type conversion
-        converted_args = []
-        converted_kwargs = kwargs.copy()
-        
-        # Handle position argument type conversion
-        for arg in args:
-            if isinstance(arg, str):
-                # Try to convert to numeric type
-                try:
-                    if '.' in arg:
-                        converted_args.append(float(arg))
-                    else:
-                        converted_args.append(int(arg))
-                except ValueError:
-                    converted_args.append(arg)
-            else:
-                converted_args.append(arg)
-        
-        # Handle keyword argument type conversion
-        for key, value in converted_kwargs.items():
-            if isinstance(value, str):
-                try:
-                    if '.' in value:
-                        converted_kwargs[key] = float(value)
-                    else:
-                        converted_kwargs[key] = int(value)
-                except ValueError:
-                    pass
-        
-        instance = YahooDownload(*converted_args, **converted_kwargs)
-        return {"success": True, "result": str(instance), "error": None}
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-@mcp.tool(name="iqfeedtool", description="IQFeedTool class")
-def iqfeedtool(*args, **kwargs):
-    """IQFeedTool class"""
-    try:
-        if IQFeedTool is None:
-            return {"success": False, "result": None, "error": "Class IQFeedTool is not available, path may need adjustment"}
-        
-        # MCP parameter type conversion
-        converted_args = []
-        converted_kwargs = kwargs.copy()
-        
-        # Handle position argument type conversion
-        for arg in args:
-            if isinstance(arg, str):
-                # Try to convert to numeric type
-                try:
-                    if '.' in arg:
-                        converted_args.append(float(arg))
-                    else:
-                        converted_args.append(int(arg))
-                except ValueError:
-                    converted_args.append(arg)
-            else:
-                converted_args.append(arg)
-        
-        # Handle keyword argument type conversion
-        for key, value in converted_kwargs.items():
-            if isinstance(value, str):
-                try:
-                    if '.' in value:
-                        converted_kwargs[key] = float(value)
-                    else:
-                        converted_kwargs[key] = int(value)
-                except ValueError:
-                    pass
-        
-        instance = IQFeedTool(*converted_args, **converted_kwargs)
-        return {"success": True, "result": str(instance), "error": None}
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-@mcp.tool(name="influxdbtool", description="InfluxDBTool class")
-def influxdbtool(*args, **kwargs):
-    """InfluxDBTool class"""
-    try:
-        if InfluxDBTool is None:
-            return {"success": False, "result": None, "error": "Class InfluxDBTool is not available, path may need adjustment"}
-        
-        # MCP parameter type conversion
-        converted_args = []
-        converted_kwargs = kwargs.copy()
-        
-        # Handle position argument type conversion
-        for arg in args:
-            if isinstance(arg, str):
-                # Try to convert to numeric type
-                try:
-                    if '.' in arg:
-                        converted_args.append(float(arg))
-                    else:
-                        converted_args.append(int(arg))
-                except ValueError:
-                    converted_args.append(arg)
-            else:
-                converted_args.append(arg)
-        
-        # Handle keyword argument type conversion
-        for key, value in converted_kwargs.items():
-            if isinstance(value, str):
-                try:
-                    if '.' in value:
-                        converted_kwargs[key] = float(value)
-                    else:
-                        converted_kwargs[key] = int(value)
-                except ValueError:
-                    pass
-        
-        instance = InfluxDBTool(*converted_args, **converted_kwargs)
-        return {"success": True, "result": str(instance), "error": None}
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-@mcp.tool(name="parse_args", description="Auto-wrapped function parse_args")
-def parse_args(payload: dict):
-    try:
-        if parse_args is None:
-            return {"success": False, "result": None, "error": "Function parse_args is not available"}
-        result = parse_args(**payload)
-        return {"success": True, "result": result, "error": None}
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-@mcp.tool(name="runstrategy", description="Auto-wrapped function runstrategy")
-def runstrategy(payload: dict):
+try:
+    import backtrader.indicators as btind
+except Exception:
+    btind = None
+
+try:
+    import backtrader.analyzers as btanalyzers
+except Exception:
+    btanalyzers = None
+
+try:
+    import backtrader.sizers as btsizers
+except Exception:
+    btsizers = None
+
+try:
+    from adapter import Adapter
+except Exception:
+    Adapter = None
+
+
+class _FallbackMCP:
+    def __init__(self, name: str) -> None:
+        self.name = name
+        self.tools: list[Any] = []
+
+    def tool(self, name: str, description: str):
+        def decorator(func):
+            setattr(func, "name", name)
+            setattr(func, "description", description)
+            self.tools.append(func)
+            return func
+
+        return decorator
+
+    def run(self, transport: str | None = None, host: str | None = None, port: int | None = None) -> None:
+        raise RuntimeError(
+            "fastmcp is not available. Install dependencies and retry. "
+            f"Requested transport={transport}, host={host}, port={port}"
+        )
+
+
+mcp = FastMCP("backtrader-mcp") if FastMCP is not None else _FallbackMCP("backtrader-mcp")
+_ADAPTER = Adapter("backtrader") if Adapter is not None else None
+
+
+def _response(success: bool, result: Any = None, error: str | None = None) -> dict[str, Any]:
+    return {"success": success, "result": result, "error": error}
+
+
+def _resolve_data_path(data_path: str) -> Path:
+    candidate = Path(data_path)
+    if candidate.is_absolute() and candidate.exists():
+        return candidate
+
+    local_candidate = _REPO_DIR / data_path
+    if local_candidate.exists():
+        return local_candidate
+
+    sample_candidate = _SOURCE_DIR.parent / "datas" / data_path
+    if sample_candidate.exists():
+        return sample_candidate
+
+    return candidate
+
+
+@mcp.tool(name="health_check", description="检查 MCP 服务和关键依赖的可用性")
+def health_check() -> dict[str, Any]:
+    """返回服务健康状态及依赖加载情况。"""
+    dependencies = {
+        "fastmcp": FastMCP is not None,
+        "backtrader": bt is not None,
+        "backtrader.feeds": btfeeds is not None,
+        "backtrader.indicators": btind is not None,
+        "backtrader.analyzers": btanalyzers is not None,
+        "backtrader.sizers": btsizers is not None,
+        "adapter": _ADAPTER is not None,
+        "source_dir_exists": _SOURCE_DIR.exists(),
+    }
+
+    adapter_health = _ADAPTER.health() if _ADAPTER is not None else {"status": "fallback", "error": "adapter unavailable"}
+    return _response(True, {"dependencies": dependencies, "adapter": adapter_health}, None)
+
+
+@mcp.tool(name="list_modules", description="列出 backtrader 已加载与加载失败的模块")
+def list_modules() -> dict[str, Any]:
+    """列出适配器当前可见的模块集合。"""
+    if _ADAPTER is None:
+        return _response(False, None, "adapter unavailable")
+
+    result = _ADAPTER.list_modules()
+    ok = result.get("status") in {"ok", "fallback"}
+    return _response(ok, result, None if ok else result.get("error", "list_modules failed"))
+
+
+@mcp.tool(name="list_symbols", description="查看指定模块的符号清单")
+def list_symbols(module_name: str, include_private: bool = False, limit: int = 200) -> dict[str, Any]:
+    """参数:
+    - module_name: 目标模块全名，例如 backtrader.indicators
+    - include_private: 是否包含以下划线开头符号
+    - limit: 返回符号数量上限
+    """
+    if _ADAPTER is None:
+        return _response(False, None, "adapter unavailable")
+
+    result = _ADAPTER.list_symbols(module_name=module_name, include_private=include_private, limit=limit)
+    ok = result.get("status") == "ok"
+    return _response(ok, result if ok else None, None if ok else result.get("error", "list_symbols failed"))
+
+
+@mcp.tool(name="preview_csv_data", description="预览本地 CSV 行情数据样本")
+def preview_csv_data(data_path: str, delimiter: str = ",", max_rows: int = 5) -> dict[str, Any]:
+    """参数:
+    - data_path: CSV 文件路径，可为绝对路径或相对路径
+    - delimiter: 分隔符，默认逗号
+    - max_rows: 最多返回行数
+    """
+    resolved = _resolve_data_path(data_path)
+    if not resolved.exists():
+        return _response(False, None, f"data file not found: {resolved}")
+
+    rows: list[list[str]] = []
     try:
-        if runstrategy is None:
-            return {"success": False, "result": None, "error": "Function runstrategy is not available"}
-        result = runstrategy(**payload)
-        return {"success": True, "result": result, "error": None}
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-@mcp.tool(name="pairtradingstrategy", description="PairTradingStrategy class")
-def pairtradingstrategy(*args, **kwargs):
-    """PairTradingStrategy class"""
+        with resolved.open("r", encoding="utf-8", newline="") as handle:
+            reader = csv.reader(handle, delimiter=delimiter)
+            for index, row in enumerate(reader):
+                rows.append(row)
+                if index + 1 >= max(max_rows, 1):
+                    break
+    except Exception as exc:
+        return _response(False, None, str(exc))
+
+    return _response(True, {"path": str(resolved), "rows": rows}, None)
+
+
+@mcp.tool(name="create_cerebro_session", description="创建并摘要一个基础 Cerebro 回测会话")
+def create_cerebro_session(initial_cash: float = 10000.0, commission: float = 0.001) -> dict[str, Any]:
+    """参数:
+    - initial_cash: 初始资金
+    - commission: 佣金比例
+    """
+    if bt is None:
+        return _response(False, None, "backtrader not available")
+
     try:
-        if PairTradingStrategy is None:
-            return {"success": False, "result": None, "error": "Class PairTradingStrategy is not available, path may need adjustment"}
-        
-        # MCP parameter type conversion
-        converted_args = []
-        converted_kwargs = kwargs.copy()
-        
-        # Handle position argument type conversion
-        for arg in args:
-            if isinstance(arg, str):
-                # Try to convert to numeric type
-                try:
-                    if '.' in arg:
-                        converted_args.append(float(arg))
-                    else:
-                        converted_args.append(int(arg))
-                except ValueError:
-                    converted_args.append(arg)
-            else:
-                converted_args.append(arg)
-        
-        # Handle keyword argument type conversion
-        for key, value in converted_kwargs.items():
-            if isinstance(value, str):
-                try:
-                    if '.' in value:
-                        converted_kwargs[key] = float(value)
-                    else:
-                        converted_kwargs[key] = int(value)
-                except ValueError:
-                    pass
-        
-        instance = PairTradingStrategy(*converted_args, **converted_kwargs)
-        return {"success": True, "result": str(instance), "error": None}
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-@mcp.tool(name="weekdaysfiller", description="WeekDaysFiller class")
-def weekdaysfiller(*args, **kwargs):
-    """WeekDaysFiller class"""
+        cerebro = bt.Cerebro()
+        cerebro.broker.setcash(float(initial_cash))
+        cerebro.broker.setcommission(commission=float(commission))
+
+        result = {
+            "cash": cerebro.broker.getcash(),
+            "commission": commission,
+            "strategy_count": len(getattr(cerebro, "strats", [])),
+            "data_count": len(getattr(cerebro, "datas", [])),
+        }
+        return _response(True, result, None)
+    except Exception as exc:
+        return _response(False, None, str(exc))
+
+
+@mcp.tool(name="run_sma_crossover_backtest", description="运行简单均线交叉策略回测")
+def run_sma_crossover_backtest(
+    data_path: str,
+    fast_period: int = 10,
+    slow_period: int = 30,
+    initial_cash: float = 10000.0,
+    commission: float = 0.001,
+    stake: int = 1,
+) -> dict[str, Any]:
+    """参数:
+    - data_path: CSV 数据路径，需包含 OHLCV
+    - fast_period: 快均线周期
+    - slow_period: 慢均线周期
+    - initial_cash: 初始资金
+    - commission: 佣金比例
+    - stake: 每次交易手数
+    """
+    if bt is None or btfeeds is None:
+        return _response(False, None, "backtrader dependencies are unavailable")
+
+    if fast_period <= 0 or slow_period <= 0:
+        return _response(False, None, "period must be positive")
+
+    if fast_period >= slow_period:
+        return _response(False, None, "fast_period must be smaller than slow_period")
+
+    resolved = _resolve_data_path(data_path)
+    if not resolved.exists():
+        return _response(False, None, f"data file not found: {resolved}")
+
+    class SmaCrossStrategy(bt.Strategy):
+        params = (("fast_period", 10), ("slow_period", 30), ("stake", 1))
+
+        def __init__(self) -> None:
+            sma_fast = bt.indicators.SMA(self.datas[0].close, period=int(self.p.fast_period))
+            sma_slow = bt.indicators.SMA(self.datas[0].close, period=int(self.p.slow_period))
+            self.crossover = bt.indicators.CrossOver(sma_fast, sma_slow)
+
+        def next(self) -> None:
+            if not self.position and self.crossover > 0:
+                self.buy(size=int(self.p.stake))
+            elif self.position and self.crossover < 0:
+                self.close()
+
     try:
-        if WeekDaysFiller is None:
-            return {"success": False, "result": None, "error": "Class WeekDaysFiller is not available, path may need adjustment"}
-        
-        # MCP parameter type conversion
-        converted_args = []
-        converted_kwargs = kwargs.copy()
-        
-        # Handle position argument type conversion
-        for arg in args:
-            if isinstance(arg, str):
-                # Try to convert to numeric type
-                try:
-                    if '.' in arg:
-                        converted_args.append(float(arg))
-                    else:
-                        converted_args.append(int(arg))
-                except ValueError:
-                    converted_args.append(arg)
-            else:
-                converted_args.append(arg)
-        
-        # Handle keyword argument type conversion
-        for key, value in converted_kwargs.items():
-            if isinstance(value, str):
-                try:
-                    if '.' in value:
-                        converted_kwargs[key] = float(value)
-                    else:
-                        converted_kwargs[key] = int(value)
-                except ValueError:
-                    pass
-        
-        instance = WeekDaysFiller(*converted_args, **converted_kwargs)
-        return {"success": True, "result": str(instance), "error": None}
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-@mcp.tool(name="parse_args", description="Auto-wrapped function parse_args")
-def parse_args(payload: dict):
+        data_feed = btfeeds.GenericCSVData(
+            dataname=str(resolved),
+            dtformat="%Y-%m-%d",
+            datetime=0,
+            open=1,
+            high=2,
+            low=3,
+            close=4,
+            volume=5,
+            openinterest=-1,
+            headers=False,
+        )
+
+        cerebro = bt.Cerebro()
+        cerebro.adddata(data_feed)
+        cerebro.addstrategy(
+            SmaCrossStrategy,
+            fast_period=int(fast_period),
+            slow_period=int(slow_period),
+            stake=int(stake),
+        )
+        cerebro.broker.setcash(float(initial_cash))
+        cerebro.broker.setcommission(commission=float(commission))
+
+        start_value = cerebro.broker.getvalue()
+        cerebro.run()
+        end_value = cerebro.broker.getvalue()
+
+        result = {
+            "data_path": str(resolved),
+            "start_value": start_value,
+            "end_value": end_value,
+            "pnl": end_value - start_value,
+            "fast_period": fast_period,
+            "slow_period": slow_period,
+        }
+        return _response(True, result, None)
+    except Exception as exc:
+        return _response(False, None, str(exc))
+
+
+@mcp.tool(name="list_builtin_components", description="列出内置指标/分析器/仓位管理组件")
+def list_builtin_components(component_type: str = "indicators", limit: int = 50) -> dict[str, Any]:
+    """参数:
+    - component_type: 可选 indicators / analyzers / sizers
+    - limit: 返回数量上限
+    """
+    if limit <= 0:
+        return _response(False, None, "limit must be positive")
+
+    mapping: dict[str, Any] = {
+        "indicators": btind,
+        "analyzers": btanalyzers,
+        "sizers": btsizers,
+    }
+
+    module = mapping.get(component_type)
+    if module is None:
+        return _response(False, None, f"unsupported component_type: {component_type}")
+
     try:
-        if parse_args is None:
-            return {"success": False, "result": None, "error": "Function parse_args is not available"}
-        result = parse_args(**payload)
-        return {"success": True, "result": result, "error": None}
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-@mcp.tool(name="runstrat", description="Auto-wrapped function runstrat")
-def runstrat(payload: dict):
+        names = [name for name in dir(module) if not name.startswith("_")]
+        sample = names[:limit]
+        return _response(True, {"component_type": component_type, "count": len(sample), "items": sample}, None)
+    except Exception as exc:
+        return _response(False, None, str(exc))
+
+
+@mcp.tool(name="adapter_call_function", description="通过适配器调用已加载模块中的函数")
+def adapter_call_function(
+    module_name: str,
+    function_name: str,
+    positional_args_json: str = "[]",
+    keyword_args_json: str = "{}",
+) -> dict[str, Any]:
+    """参数:
+    - module_name: 模块名
+    - function_name: 函数名
+    - positional_args_json: JSON 数组字符串
+    - keyword_args_json: JSON 对象字符串
+    """
+    if _ADAPTER is None:
+        return _response(False, None, "adapter unavailable")
+
     try:
-        if runstrat is None:
-            return {"success": False, "result": None, "error": "Function runstrat is not available"}
-        result = runstrat(**payload)
-        return {"success": True, "result": result, "error": None}
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
+        positional_args = json.loads(positional_args_json)
+        keyword_args = json.loads(keyword_args_json)
+    except Exception as exc:
+        return _response(False, None, f"invalid JSON args: {exc}")
+
+    if not isinstance(positional_args, list):
+        return _response(False, None, "positional_args_json must decode to a list")
+    if not isinstance(keyword_args, dict):
+        return _response(False, None, "keyword_args_json must decode to an object")
 
+    result = _ADAPTER.call_function(
+        module_name=module_name,
+        function_name=function_name,
+        positional_args=positional_args,
+        keyword_args=keyword_args,
+    )
 
+    ok = result.get("status") == "ok"
+    return _response(ok, result if ok else None, None if ok else result.get("error", "adapter call failed"))
 
-def create_app():
-    """Create and return FastMCP application instance"""
+
+def create_app() -> Any:
     return mcp
 
+
 if __name__ == "__main__":
-    mcp.run(transport="http", host="0.0.0.0", port=8000)
+    mcp.run()

backtrader/mcp_output/requirements.txt

@@ -1,6 +1,2 @@
 fastmcp
-fastapi
-uvicorn[standard]
-pydantic>=2.0.0
-six
-matplotlib
+backtrader

backtrader/mcp_output/start_mcp.py

@@ -1,30 +1,35 @@
+from __future__ import annotations
 
-"""
-MCP Service Startup Entry
-"""
-import sys
 import os
+import sys
+from pathlib import Path
+
+CURRENT_DIR = Path(__file__).resolve().parent
+PLUGIN_DIR = CURRENT_DIR / "mcp_plugin"
 
-project_root = os.path.dirname(os.path.abspath(__file__))
-mcp_plugin_dir = os.path.join(project_root, "mcp_plugin")
-if mcp_plugin_dir not in sys.path:
-    sys.path.insert(0, mcp_plugin_dir)
+if str(PLUGIN_DIR) not in sys.path:
+    sys.path.insert(0, str(PLUGIN_DIR))
 
 from mcp_service import create_app
 
-def main():
-    """Start FastMCP service"""
+
+def main() -> None:
+    transport = os.getenv("MCP_TRANSPORT", "stdio").strip().lower()
+    port_raw = os.getenv("MCP_PORT", "8000").strip()
+
+    try:
+        port = int(port_raw)
+    except ValueError:
+        port = 8000
+
     app = create_app()
-    # Use environment variable to configure port, default 8000
-    port = int(os.environ.get("MCP_PORT", "8000"))
-    
-    # Choose transport mode based on environment variable
-    transport = os.environ.get("MCP_TRANSPORT", "stdio")
+
     if transport == "http":
         app.run(transport="http", host="0.0.0.0", port=port)
-    else:
-        # Default to STDIO mode
-        app.run()
+        return
+
+    app.run(transport="stdio")
+
 
 if __name__ == "__main__":
     main()

port.json

@@ -1,5 +1 @@
-{
-  "repo": "backtrader",
-  "port": 7946,
-  "timestamp": 1773378088
-}
+{"port": 7860}

requirements.txt

@@ -1,6 +1,4 @@
 fastmcp
+backtrader
 fastapi
-uvicorn[standard]
-pydantic>=2.0.0
-six
-matplotlib
+uvicorn

run_docker.ps1

@@ -1,26 +1,10 @@
-cd $PSScriptRoot
 $ErrorActionPreference = "Stop"
-$entryName = if ($env:MCP_ENTRY_NAME) { $env:MCP_ENTRY_NAME } else { "backtrader" }
-$entryUrl  = if ($env:MCP_ENTRY_URL)  { $env:MCP_ENTRY_URL  } else { "http://localhost:7946/mcp" }
-$imageName = if ($env:MCP_IMAGE_NAME) { $env:MCP_IMAGE_NAME } else { "backtrader-mcp" }
-$mcpDir = Join-Path $env:USERPROFILE ".cursor"
-$mcpPath = Join-Path $mcpDir "mcp.json"
-if (!(Test-Path $mcpDir)) { New-Item -ItemType Directory -Path $mcpDir | Out-Null }
-$config = @{}
-if (Test-Path $mcpPath) {
-  try { $config = Get-Content $mcpPath -Raw | ConvertFrom-Json } catch { $config = @{} }
-}
-$serversOrdered = [ordered]@{}
-if ($config -and ($config.PSObject.Properties.Name -contains "mcpServers") -and $config.mcpServers) {
-  $existing = $config.mcpServers
-  if ($existing -is [pscustomobject]) {
-    foreach ($p in $existing.PSObject.Properties) { if ($p.Name -ne $entryName) { $serversOrdered[$p.Name] = $p.Value } }
-  } elseif ($existing -is [System.Collections.IDictionary]) {
-    foreach ($k in $existing.Keys) { if ($k -ne $entryName) { $serversOrdered[$k] = $existing[$k] } }
-  }
-}
-$serversOrdered[$entryName] = @{ url = $entryUrl }
-$config = @{ mcpServers = $serversOrdered }
-$config | ConvertTo-Json -Depth 10 | Set-Content -Path $mcpPath -Encoding UTF8
-docker build -t $imageName .
-docker run --rm -p 7946:7860 $imageName
+
+$ScriptDir = Split-Path -Parent $MyInvocation.MyCommand.Path
+$PortConfig = Get-Content (Join-Path $ScriptDir "port.json") -Raw | ConvertFrom-Json
+$Port = [int]$PortConfig.port
+$ImageName = "backtrader-mcp"
+
+Set-Location $ScriptDir
+docker build -t $ImageName .
+docker run --rm -p "${Port}:7860" $ImageName

run_docker.sh

@@ -1,75 +1,10 @@
 #!/usr/bin/env bash
 set -euo pipefail
-cd "$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)"
-mcp_entry_name="${MCP_ENTRY_NAME:-backtrader}"
-mcp_entry_url="${MCP_ENTRY_URL:-http://localhost:7946/mcp}"
-mcp_dir="${HOME}/.cursor"
-mcp_path="${mcp_dir}/mcp.json"
-mkdir -p "${mcp_dir}"
-if command -v python3 >/dev/null 2>&1; then
-python3 - "${mcp_path}" "${mcp_entry_name}" "${mcp_entry_url}" <<'PY'
-import json, os, sys
-path, name, url = sys.argv[1:4]
-cfg = {"mcpServers": {}}
-if os.path.exists(path):
-    try:
-        with open(path, "r", encoding="utf-8") as f:
-            cfg = json.load(f)
-    except Exception:
-        cfg = {"mcpServers": {}}
-if not isinstance(cfg, dict):
-    cfg = {"mcpServers": {}}
-servers = cfg.get("mcpServers")
-if not isinstance(servers, dict):
-    servers = {}
-ordered = {}
-for k, v in servers.items():
-    if k != name:
-        ordered[k] = v
-ordered[name] = {"url": url}
-cfg = {"mcpServers": ordered}
-with open(path, "w", encoding="utf-8") as f:
-    json.dump(cfg, f, indent=2, ensure_ascii=False)
-PY
-elif command -v python >/dev/null 2>&1; then
-python - "${mcp_path}" "${mcp_entry_name}" "${mcp_entry_url}" <<'PY'
-import json, os, sys
-path, name, url = sys.argv[1:4]
-cfg = {"mcpServers": {}}
-if os.path.exists(path):
-    try:
-        with open(path, "r", encoding="utf-8") as f:
-            cfg = json.load(f)
-    except Exception:
-        cfg = {"mcpServers": {}}
-if not isinstance(cfg, dict):
-    cfg = {"mcpServers": {}}
-servers = cfg.get("mcpServers")
-if not isinstance(servers, dict):
-    servers = {}
-ordered = {}
-for k, v in servers.items():
-    if k != name:
-        ordered[k] = v
-ordered[name] = {"url": url}
-cfg = {"mcpServers": ordered}
-with open(path, "w", encoding="utf-8") as f:
-    json.dump(cfg, f, indent=2, ensure_ascii=False)
-PY
-elif command -v jq >/dev/null 2>&1; then
-  name="${mcp_entry_name}"; url="${mcp_entry_url}"
-  if [ -f "${mcp_path}" ]; then
-    tmp="$(mktemp)"
-    jq --arg name "$name" --arg url "$url" '
-      .mcpServers = (.mcpServers // {})
-      | .mcpServers as $s
-      | ($s | with_entries(select(.key != $name))) as $base
-      | .mcpServers = ($base + {($name): {"url": $url}})
-    ' "${mcp_path}" > "${tmp}" && mv "${tmp}" "${mcp_path}"
-  else
-    printf '{ "mcpServers": { "%s": { "url": "%s" } } }
-' "$name" "$url" > "${mcp_path}"
-  fi
-fi
-docker build -t backtrader-mcp .
-docker run --rm -p 7946:7860 backtrader-mcp
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+PORT="$(python3 -c 'import json,sys;print(json.load(open(sys.argv[1]))["port"])' "$SCRIPT_DIR/port.json")"
+IMAGE_NAME="backtrader-mcp"
+
+cd "$SCRIPT_DIR"
+docker build -t "$IMAGE_NAME" .
+docker run --rm -p "${PORT}:7860" "$IMAGE_NAME"