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

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 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/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/tradingagents/dataflows/data_source_manager.py)
Interface	Public façade (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/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 (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:

# 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:

# 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 so the manager can discover and instantiate it.

Update the Enum

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

# 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:

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

{
  "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).

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.

Automatic Fallback Behavior

When a request is sent through 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).

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

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	Abstract base class that all providers must implement.	[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).	Create under the appropriate sub‑folder
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/tradingagents/dataflows/data_source_manager.py)
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/tradingagents/dataflows/interface.py)
constants.py (or similar)	Holds DataSourceCode values referenced by enums.	[constants.py](https://github.com/hsliuping/TradingAgents-CN/blob/main/tradingagents/constants.py)
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/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 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 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) and ensure it contains both the provider class and the factory function. This structure allows DataSourceManager to discover and import your module cleanly.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}