# How to Integrate Additional Data Sources into the TradingAgents-CN Dataflows System

> Learn to integrate new data sources into TradingAgents-CN. Implement a provider class, register it, and enable via MongoDB for seamless dataflow management.

- Repository: [hsliuping/TradingAgents-CN](https://github.com/hsliuping/tradingagents-cn)
- Tags: how-to-guide
- Published: 2026-02-16

---

**To integrate a new data source in TradingAgents-CN, implement a provider class inheriting from `BaseStockDataProvider`, register it in the `DataSourceManager` enum, and enable it via the MongoDB `system_configs` collection.**

The **dataflows** subsystem is the backbone that supplies market data—prices, fundamentals, and news—to agents, APIs, and the web UI in the [TradingAgents-CN](https://github.com/hsliuping/TradingAgents-CN) repository. Adding a new data source requires no changes to core logic or agent code; the system automatically discovers and falls back to your provider using a priority-based registry.

## Understanding the Dataflows Architecture

The system is built around three core concepts that work together to abstract data retrieval:

| Concept | Description | Core Files |
|---|---|---|
| **Provider** | A concrete implementation that fetches raw data from a third-party service (e.g., Tushare, AKShare, Yahoo Finance). All providers inherit from `BaseStockDataProvider`, which defines a common async interface. | [[`providers/base_provider.py`](https://github.com/hsliuping/TradingAgents-CN/blob/main/providers/base_provider.py)](https://github.com/hsliuping/TradingAgents-CN/blob/main/tradingagents/dataflows/providers/base_provider.py) |
| **DataSourceManager** | Central registry that discovers installed providers, checks which are enabled in the database configuration, and decides the priority order for automatic fallback. | [[`dataflows/data_source_manager.py`](https://github.com/hsliuping/TradingAgents-CN/blob/main/dataflows/data_source_manager.py)](https://github.com/hsliuping/TradingAgents-CN/blob/main/tradingagents/dataflows/data_source_manager.py) |
| **Interface** | Public façade ([`interface.py`](https://github.com/hsliuping/TradingAgents-CN/blob/main/interface.py)) that agents and external callers use. It delegates requests to the manager, which picks the appropriate provider or falls back to the next available source. | [[`dataflows/interface.py`](https://github.com/hsliuping/TradingAgents-CN/blob/main/dataflows/interface.py)](https://github.com/hsliuping/TradingAgents-CN/blob/main/tradingagents/dataflows/interface.py) |

## Step 1: Implementing the Provider Class

Every provider must derive from `BaseStockDataProvider` and implement three core async methods. The base class lives in [`providers/base_provider.py`](https://github.com/hsliuping/TradingAgents-CN/blob/main/providers/base_provider.py) (lines 11–59), defining the interface and standardization helpers.

### Required Implementation Steps

1. **Inherit from `BaseStockDataProvider`** – This ensures the provider implements `connect()`, `get_stock_basic_info()`, `get_stock_quotes()`, and `get_historical_data()`.
2. **Implement the three core methods** – These should return raw data from the third-party API.
3. **Standardize output** – After fetching, call the `standardize_*` helpers from the base class so downstream code receives a uniform dictionary or DataFrame.

### Example Provider Implementation

The following skeleton demonstrates a fictional "MyFinance" provider for US equities:

```python

# tradingagents/dataflows/providers/us/myfinance.py

from .base_provider import BaseStockDataProvider
import httpx
import pandas as pd

class MyFinanceProvider(BaseStockDataProvider):
    async def connect(self) -> bool:
        # No persistent connection needed for a REST API

        self.connected = True
        return True

    async def get_stock_basic_info(self, symbol: str = None):
        # Return a list of dicts with basic info

        url = "https://api.myfinance.com/v1/stocks/info"
        params = {"symbol": symbol} if symbol else {}
        async with httpx.AsyncClient() as client:
            resp = await client.get(url, params=params)
            data = resp.json()
        return [self.standardize_basic_info(item) for item in data["items"]]

    async def get_stock_quotes(self, symbol: str):
        url = f"https://api.myfinance.com/v1/quote/{symbol}"
        async with httpx.AsyncClient() as client:
            resp = await client.get(url)
            raw = resp.json()
        return self.standardize_quotes(raw)

    async def get_historical_data(self, symbol, start_date, end_date=None):
        url = f"https://api.myfinance.com/v1/history/{symbol}"
        params = {"start": start_date, "end": end_date}
        async with httpx.AsyncClient() as client:
            resp = await client.get(url, params=params)
            df = pd.DataFrame(resp.json()["records"])
        return self._standardize_dataframe(df)

```

Expose a factory function for the manager to consume:

```python

# factory function used by the manager

def get_myfinance_provider() -> MyFinanceProvider:
    # a single shared instance is sufficient

    if not hasattr(get_myfinance_provider, "_instance"):
        get_myfinance_provider._instance = MyFinanceProvider("myfinance")
    return get_myfinance_provider._instance

```

Place the file under `tradingagents/dataflows/providers/us/` (or `china/` for A-share sources) and commit it to your branch.

## Step 2: Registering the Provider with DataSourceManager

Once the provider class exists, you must register it in [`dataflows/data_source_manager.py`](https://github.com/hsliuping/TradingAgents-CN/blob/main/dataflows/data_source_manager.py) so the manager can discover and instantiate it.

### Update the Enum

Add a new entry to the appropriate market enum (around line 41):

```python

# Add a new enum entry (around line 41)

class USDataSource(Enum):
    MONGODB = DataSourceCode.MONGODB
    YFINANCE = DataSourceCode.YFINANCE
    ALPHA_VANTAGE = DataSourceCode.ALPHA_VANTAGE
    FINNHUB = DataSourceCode.FINNHUB
    MYFINANCE = DataSourceCode.MYFINANCE   # ← new entry

```

Ensure the value `MYFINANCE` exists in `tradingagents.constants.DataSourceCode`.

### Update the Availability Check

Modify `_check_available_sources` (around lines 64–98) to import and test your provider:

```python
if 'myfinance' in enabled_sources_in_db:
    try:
        import httpx  # verify dependency

        from .providers.us.myfinance import get_myfinance_provider
        available.append(USDataSource.MYFINANCE)
        logger.info("✅ MyFinance 数据源可用且已启用")
    except Exception as e:
        logger.warning(f"⚠️ MyFinance 数据源不可用: {e}")

```

The manager will now include your provider in `self.available_sources` if the database flag is set and dependencies are importable.

## Step 3: Enabling the Source via Database Configuration

Data source activation is controlled by the **system config collection** in MongoDB (`system_configs`). Each document defines whether a provider is active and its fallback priority.

### Configuration Document Structure

```json
{
  "type": "myfinance",
  "enabled": true,
  "priority": 30,
  "market_categories": ["us_stocks"]
}

```

### Configuration Steps

1. **Insert or update a document** in `system_configs` with `type` matching the lower-case name used in your factory function (`myfinance`).
2. **Set `enabled`** to `true` to allow the manager to load it.
3. **Assign a `priority`** value—higher numbers take precedence when the manager selects a source or falls back.
4. **Optionally restrict** it to specific markets using `market_categories` (e.g., `["a_shares"]` for China, `["us_stocks"]` for US).

The manager reads this configuration in `_get_datasource_configs_from_db()` (around line 1090 of [`data_source_manager.py`](https://github.com/hsliuping/TradingAgents-CN/blob/main/data_source_manager.py)).

## Step 4: Accessing Data Through the Interface

Once registered and enabled, agents consume the new provider transparently through the public façade in [`dataflows/interface.py`](https://github.com/hsliuping/TradingAgents-CN/blob/main/dataflows/interface.py).

### Automatic Fallback Behavior

When a request is sent through [`interface.py`](https://github.com/hsliuping/TradingAgents-CN/blob/main/interface.py), the manager first tries the **current source** (which may be forced by the user). If the call fails or returns malformed data, the manager iterates over `self.available_sources` (populated by `_check_available_sources`) and attempts each provider in order of priority (see the loop starting at line 450 of [`data_source_manager.py`](https://github.com/hsliuping/TradingAgents-CN/blob/main/data_source_manager.py)).

Therefore, once the new provider is correctly registered and marked as enabled, it will automatically participate in fallback without any further code changes.

### Example Usage from an Agent

```python
from tradingagents.dataflows.interface import set_config, get_us_stock_data_unified

# Optional: force the manager to use MyFinance for the next call

set_config({"default_us_data_source": "myfinance"})   # internal helper

data = get_us_stock_data_unified(symbol="AAPL", start_date="2024-01-01", end_date="2024-02-01")
print(data)   # receives a formatted string from the provider

```

If `MyFinance` fails, the manager will automatically fall back to the next enabled source (e.g., YFinance) thanks to the fallback loop in `DataSourceManager.get_stock_data()`.

## Key Files and Their Roles

| File | Role | Link |
|---|---|---|
| [`providers/base_provider.py`](https://github.com/hsliuping/TradingAgents-CN/blob/main/providers/base_provider.py) | Abstract base class that all providers must implement. | [[`base_provider.py`](https://github.com/hsliuping/TradingAgents-CN/blob/main/base_provider.py)](https://github.com/hsliuping/TradingAgents-CN/blob/main/tradingagents/dataflows/providers/base_provider.py) |
| `providers/<market>/<your_provider>.py` | Concrete provider implementation (e.g., [`us/myfinance.py`](https://github.com/hsliuping/TradingAgents-CN/blob/main/us/myfinance.py)). | *Create under the appropriate sub‑folder* |
| [`dataflows/data_source_manager.py`](https://github.com/hsliuping/TradingAgents-CN/blob/main/dataflows/data_source_manager.py) | Discovers enabled providers, builds priority list, handles fallback. | [[`data_source_manager.py`](https://github.com/hsliuping/TradingAgents-CN/blob/main/data_source_manager.py)](https://github.com/hsliuping/TradingAgents-CN/blob/main/tradingagents/dataflows/data_source_manager.py) |
| [`dataflows/interface.py`](https://github.com/hsliuping/TradingAgents-CN/blob/main/dataflows/interface.py) | Public façade used by agents, CLI, web API. Delegates to the manager. | [[`interface.py`](https://github.com/hsliuping/TradingAgents-CN/blob/main/interface.py)](https://github.com/hsliuping/TradingAgents-CN/blob/main/tradingagents/dataflows/interface.py) |
| [`constants.py`](https://github.com/hsliuping/TradingAgents-CN/blob/main/constants.py) (or similar) | Holds `DataSourceCode` values referenced by enums. | [[`constants.py`](https://github.com/hsliuping/TradingAgents-CN/blob/main/constants.py)](https://github.com/hsliuping/TradingAgents-CN/blob/main/tradingagents/constants.py) |
| [`config/runtime_settings.py`](https://github.com/hsliuping/TradingAgents-CN/blob/main/config/runtime_settings.py) (MongoDB config) | Supplies DB‑backed activation flags that `DataSourceManager` reads. | [[`runtime_settings.py`](https://github.com/hsliuping/TradingAgents-CN/blob/main/runtime_settings.py)](https://github.com/hsliuping/TradingAgents-CN/blob/main/tradingagents/config/runtime_settings.py) |

These files together define the plug‑in architecture. Adding a new data source only requires **(1) a provider class**, **(2) a factory**, **(3) enum registration**, and **(4) a DB config entry**—no changes to the core logic or agents. The system will automatically leverage the new source and include it in the existing fallback chain.

## Summary

- **Inherit from `BaseStockDataProvider`** in [`providers/base_provider.py`](https://github.com/hsliuping/TradingAgents-CN/blob/main/providers/base_provider.py) to ensure your implementation follows the required async interface (`connect`, `get_stock_basic_info`, `get_stock_quotes`, `get_historical_data`).
- **Expose a singleton factory** (e.g., `get_myfinance_provider()`) so `DataSourceManager` can instantiate your provider without hard-coding imports.
- **Register the provider** by adding an enum entry to `ChinaDataSource` or `USDataSource` in [`dataflows/data_source_manager.py`](https://github.com/hsliuping/TradingAgents-CN/blob/main/dataflows/data_source_manager.py) and updating `_check_available_sources` to import and test your module.
- **Enable via MongoDB** by inserting a document into `system_configs` with `type`, `enabled: true`, a `priority` value, and optional `market_categories`.
- **Leverage automatic fallback**—once registered, your provider participates in the priority chain managed by `DataSourceManager.get_stock_data()` without further code changes.

## Frequently Asked Questions

### What is the minimum code required to add a new data source?

You need four components: a provider class inheriting from `BaseStockDataProvider` (implementing `connect`, `get_stock_basic_info`, `get_stock_quotes`, and `get_historical_data`), a factory function returning a singleton instance, an enum entry in `DataSourceManager`, and a configuration document in the MongoDB `system_configs` collection. No modifications to agent logic or the web UI are necessary.

### How does the system handle failures when my new data source is unavailable?

The `DataSourceManager` implements automatic fallback logic in `get_stock_data()` (starting around line 450). If the primary provider fails or returns malformed data, the manager iterates through `self.available_sources` in priority order (highest first) until it finds a working source. Your new provider automatically joins this fallback chain once registered and enabled.

### Can I restrict a data source to specific markets (e.g., only A-shares)?

Yes. When inserting the configuration document into MongoDB, include the `market_categories` field with an array of strings such as `["a_shares"]` for China or `["us_stocks"]` for US equities. The `DataSourceManager` uses this field to filter which providers are eligible for specific market requests, ensuring your provider is only called for relevant symbols.

### Where should I place the new provider file in the codebase?

Create a new module under the appropriate market sub-package: `tradingagents/dataflows/providers/china/` for A-share data sources or `tradingagents/dataflows/providers/us/` for US equities. Name the file after your provider (e.g., [`myfinance.py`](https://github.com/hsliuping/TradingAgents-CN/blob/main/myfinance.py)) and ensure it contains both the provider class and the factory function. This structure allows `DataSourceManager` to discover and import your module cleanly.