ljwx-stock/DATA_SOURCE_COMPARISON.md

数据源全面对比分析报告

基于实际测试结果（2025-12-21）

测试股票：600000（浦发银行）、000001（平安银行）、600519（贵州茅台）

---

📊 测试结果总览

综合评分排名

排名	数据源	综合评分	实时行情	历史K线	基本面	推荐度
🥇 1	腾讯财经	⭐⭐⭐⭐⭐	✅ 5/5	✅ 5/5	-	⭐⭐⭐⭐⭐
🥈 2	新浪财经	⭐⭐⭐⭐½	✅ 5/5	✅ 4/5	-	⭐⭐⭐⭐⭐
🥉 3	东方财富	⭐⭐⭐	✅ 5/5	❌ 0/5	✅ 4/5	⭐⭐⭐⭐
4	AkShare	⭐⭐⭐⭐⭐	✅ 5/5	✅ 5/5	✅ 5/5	⭐⭐⭐⭐⭐
5	百度股市通	⭐	⚠️ 1/5	-	-	⭐
6	同花顺	☆	❌ 0/5	❌ 0/5	-	☆
7	网易财经	☆	❌ 0/5	-	-	☆

注: AkShare 是之前测试的数据库，本次主要测试原生API接口

---

🔍 详细测试结果

1. 新浪财经 - 老牌稳定 ⭐⭐⭐⭐⭐

✅ 实时行情测试

API接口:

http://hq.sinajs.cn/list=sh600000,sz000001,sh600519

测试结果:

✅ HTTP 200 | 响应时间: 0.03s
✅ 成功获取 3 只股票数据
✅ 字段数量: 32个

返回数据示例:

var hq_str_sh600000="浦发银行,11.720,11.780,11.690,11.800,11.650,11.690,11.700,67815119,794336354.000,..."

字段说明:

索引	字段	示例值	说明
0	股票名称	浦发银行	-
1	今日开盘价	11.72	-
2	昨日收盘价	11.78	-
3	当前价格	11.69	核心字段
4	今日最高价	11.80	-
5	今日最低价	11.65	-
6	买一价	11.69	-
7	卖一价	11.70	-
8	成交量(股)	67815119	-
9	成交额(元)	794336354	-
10-17	买1-5	-	五档买盘
18-27	卖1-5	-	五档卖盘
30	日期	2025-12-19	-
31	时间	15:00:03	-

优点:

• ✅ 响应速度极快（30ms）
• ✅ 数据稳定可靠
• ✅ 无需认证
• ✅ 支持批量查询（逗号分隔）
• ✅ 字段丰富（32个字段，包含五档行情）
• ✅ 数据格式简单
• ✅ 免费无限制

缺点:

• ⚠️ 返回GBK编码文本，需要解析
• ⚠️ 字段为位置索引，不够语义化
• ⚠️ 无JSON格式

✅ 历史K线测试

API接口:

http://money.finance.sina.com.cn/quotes_service/api/json_v2.php/CN_MarketData.getKLineData?symbol=sh600000&scale=240&datalen=1023

测试结果:

✅ HTTP 200
✅ 成功获取 1023 条历史数据
✅ 数据格式: JSON数组

返回数据示例:

{
  "day": "2025-12-19",
  "open": "11.720",
  "high": "11.800",
  "low": "11.650",
  "close": "11.690",
  "volume": "67815119"
}

优点:

• ✅ JSON格式，易于解析
• ✅ 数据量大（支持1023条）
• ✅ 字段完整（OHLCV）

缺点:

• ⚠️ 接口稳定性一般
• ⚠️ 无换手率、成交额等扩展字段

总体评价: ⭐⭐⭐⭐⭐

• 最佳使用场景: 实时行情监控
• 推荐指数: 5/5

---

2. 腾讯财经 - 全能冠军 ⭐⭐⭐⭐⭐

✅ 实时行情测试

API接口:

http://qt.gtimg.cn/q=sh600000,sz000001,sh600519

测试结果:

✅ HTTP 200
✅ 成功获取 3 只股票数据
✅ 字段数量: 40+个

返回数据示例:

v_sh600000="1~浦发银行~600000~11.69~11.78~11.72~678151~278835~399316~11.69~6786~11.68~1805~..."

核心字段:

索引	字段	示例值
1	股票名称	浦发银行
3	当前价格	11.69
4	昨收	11.78
5	今开	11.72
6	成交量(手)	678151
7	外盘	278835
8	内盘	399316
9-13	买1-5	...
19-28	更多指标	涨跌幅、换手率等
33	最高	11.80
34	最低	11.65

优点:

• ✅ 字段最丰富（40+个字段）
• ✅ 包含内外盘数据
• ✅ 响应稳定
• ✅ 支持批量查询
• ✅ 免费无限制

缺点:

• ⚠️ 返回GBK文本格式
• ⚠️ 字段为波浪号分隔，需要解析

✅ 历史K线测试 - 最佳

API接口:

http://web.ifzq.gtimg.cn/appstock/app/fqkline/get?param=sh600000,day,,,320,qfq&_var=kline_dayqfq

测试结果:

✅ HTTP 200
✅ 成功获取 320 条K线数据
✅ 数据格式: JSONP包裹的JSON

返回数据示例:

kline_dayqfq={
  "data": {
    "sh600000": {
      "qfqday": [
        ["2025-12-19", "11.720", "11.690", "11.800", "11.650", "67815119"],
        ...
      ]
    }
  }
}

字段说明:

[日期, 开盘, 收盘, 最高, 最低, 成交量]

优点:

• ✅ 数据量充足（支持320条+）
• ✅ JSON格式（虽然包裹在JSONP中）
• ✅ 支持前复权/后复权/不复权
• ✅ 数据准确，来源权威
• ✅ 支持多种周期（日/周/月/分钟）

缺点:

• ⚠️ JSONP格式需要去掉前缀
• ⚠️ 需要了解参数含义

参数说明:

param=sh600000,day,,,320,qfq
       ↓        ↓       ↓   ↓
       股票代码  周期  数量  复权方式

周期: day(日), week(周), month(月), m5(5分钟), m15, m30, m60
复权: qfq(前复权), hfq(后复权), (空=不复权)

总体评价: ⭐⭐⭐⭐⭐

• 最佳使用场景: 历史K线数据获取
• 推荐指数: 5/5

---

3. 东方财富 - 基本面专家 ⭐⭐⭐⭐

✅ 实时行情测试

API接口:

http://push2.eastmoney.com/api/qt/stock/get?secid=1.600000&fields=f43,f44,f45,f46,f47,f48,f168

测试结果:

✅ HTTP 200
✅ 成功获取 2 只股票数据
✅ 数据格式: 标准JSON

返回数据示例:

{
  "rc": 0,
  "data": {
    "f43": 1169,      // 最新价 * 100
    "f44": 1180,      // 最高价 * 100
    "f45": 1165,      // 最低价 * 100
    "f46": 1172,      // 开盘价 * 100
    "f47": 678151,    // 成交量(手)
    "f48": 794336354, // 成交额
    "f58": "浦发银行",
    "f168": 20        // 换手率 * 100
  }
}

字段映射:

字段代码	含义	注意事项
f43	最新价	需除以100
f44	最高价	需除以100
f45	最低价	需除以100
f46	开盘价	需除以100
f47	成交量	单位：手
f48	成交额	单位：元
f57	股票代码	-
f58	股票名称	-
f168	换手率	需除以100，单位%

优点:

• ✅ 标准JSON格式
• ✅ 字段语义明确
• ✅ 包含换手率（新浪、腾讯不提供）
• ✅ 数据准确

缺点:

• ⚠️ 价格需除以100（容易出错）
• ⚠️ secid格式需要转换（1.600000 表示上海，0.000001表示深圳）
• ⚠️ 字段代码不直观（f43, f44...）

❌ 历史K线测试 - 失败

API接口:

http://push2his.eastmoney.com/api/qt/stock/kline/get?secid=1.600000&klt=101&fqt=1&lmt=120

测试结果:

❌ 未获取到K线数据
⚠️ 可能原因：接口参数变更或需要额外认证

说明:

• 该接口之前可用，现在返回空数据
• 可能需要Cookie或Referer认证
• 建议使用其他数据源的K线接口

✅ 基本面数据测试 - 独有优势

API接口:

http://push2.eastmoney.com/api/qt/stock/get?secid=1.600000&fields=f57,f58,f162,f167

测试结果:

✅ HTTP 200
✅ 成功获取基本面数据

返回数据:

浦发银行 基本面数据:
  市盈率(动): 7.52
  市净率: 0.54

基本面字段映射:

字段	含义	计算方式
f162	市盈率(动态)	需除以100
f167	市净率	需除以100
f116	总市值	亿元
f117	流通市值	亿元

优点:

• ✅ 唯一提供基本面数据的免费API
• ✅ 数据更新及时
• ✅ JSON格式

总体评价: ⭐⭐⭐⭐

• 最佳使用场景: 基本面数据获取、实时行情（配合换手率）
• 推荐指数: 4/5
• 注意事项: K线接口不稳定，建议使用其他数据源

---

4. 同花顺 - 不推荐 ☆

❌ 实时行情测试

API接口:

http://d.10jqka.com.cn/v2/realhead/hs_sh600000/last.js

测试结果:

❌ HTTP 404 - Not Found

❌ K线数据测试

API接口:

http://d.10jqka.com.cn/v2/line/hs_sh600000/01/last.js

测试结果:

❌ HTTP 404 - Not Found

总体评价: ☆

• 结论: 接口已失效或大幅变更
• 推荐指数: 0/5
• 建议: 不推荐使用

---

5. 百度股市通 - 参数问题 ⭐

⚠️ 实时行情测试

API接口:

https://gushitong.baidu.com/opendata?resource=Finance::Stock::RealtimeUS&query=sh600000&code=sh600000

测试结果:

✅ HTTP 200
❌ 返回 "参数错误"

返回数据:

{
  "status": 1,
  "msg": "参数错误",
  "data": []
}

总体评价: ⭐

• 问题: 参数格式可能变更或需要特殊认证
• 推荐指数: 1/5
• 建议: 需要进一步研究参数格式

---

6. 网易财经 - 连接失败 ☆

❌ 实时行情测试

API接口:

http://api.money.126.net/data/feed/0600000,1000001

测试结果:

❌ DNS解析失败
❌ 域名无法访问

总体评价: ☆

• 问题: API域名可能已停用
• 推荐指数: 0/5
• 建议: 不推荐使用

---

📈 综合对比分析

数据源特点矩阵

数据源	实时行情	历史K线	基本面	响应速度	JSON格式	免费	稳定性
腾讯财经	⭐⭐⭐⭐⭐	⭐⭐⭐⭐⭐	❌	快	⚠️ JSONP	✅	⭐⭐⭐⭐⭐
新浪财经	⭐⭐⭐⭐⭐	⭐⭐⭐⭐	❌	极快	❌ 文本	✅	⭐⭐⭐⭐⭐
东方财富	⭐⭐⭐⭐⭐	❌	⭐⭐⭐⭐⭐	中	✅	✅	⭐⭐⭐⭐
AkShare	⭐⭐⭐⭐⭐	⭐⭐⭐⭐⭐	⭐⭐⭐⭐⭐	慢	✅ DF	✅	⭐⭐⭐⭐

使用场景推荐

场景1: 实时行情展示（Dashboard）

推荐方案: 新浪财经 + 腾讯财经（双备份）

async def get_realtime_quote(symbol: str):
    """实时行情 - 双数据源容错"""

    # 优先：新浪（速度快）
    try:
        return await fetch_sina_realtime(symbol)
    except:
        pass

    # 备用：腾讯（字段全）
    try:
        return await fetch_tencent_realtime(symbol)
    except:
        raise DataUnavailableError()

优点:

• ✅ 极低延迟（30-50ms）
• ✅ 高可用（双备份）
• ✅ 字段丰富

场景2: 历史数据加载（Database）

推荐方案: 腾讯财经 K线API

async def load_historical_data(symbol: str, days: int = 365):
    """历史数据加载"""

    # 腾讯K线API - 最多支持320条
    url = "http://web.ifzq.gtimg.cn/appstock/app/fqkline/get"
    params = {
        'param': f'{symbol},day,,,{days},qfq',
        '_var': 'kline_dayqfq'
    }

    response = await fetch(url, params)
    # 解析JSONP
    json_str = response.replace('kline_dayqfq=', '')
    data = json.loads(json_str)

    return parse_kline_data(data)

优点:

• ✅ 数据量充足（320条 ≈ 1.3年）
• ✅ 支持复权
• ✅ 数据准确

补充: 如需更长历史数据，使用 AkShare

场景3: 基本面分析

推荐方案: 东方财富 + AkShare

async def get_fundamentals(symbol: str):
    """基本面数据"""

    # 东方财富 - 实时PE/PB
    market = '1' if symbol.startswith('6') else '0'
    secid = f"{market}.{symbol}"

    url = "http://push2.eastmoney.com/api/qt/stock/get"
    params = {
        'secid': secid,
        'fields': 'f162,f167,f116,f117'  # PE、PB、市值
    }

    response = await fetch(url, params)
    data = response['data']

    return {
        'pe': data['f162'] / 100,
        'pb': data['f167'] / 100,
        'market_cap': data['f116']
    }

优点:

• ✅ 实时更新
• ✅ 数据准确
• ✅ 免费

场景4: 大规模数据采集

推荐方案: AkShare

import akshare as ak

# 一次性获取所有A股实时数据
df = ak.stock_zh_a_spot_em()  # 5790只股票

# 批量加载历史数据
for symbol in stock_list:
    df = ak.stock_zh_a_hist(symbol=symbol, period="daily", adjust="qfq")
    await save_to_database(df)

优点:

• ✅ 接口封装完善
• ✅ 返回DataFrame，直接入库
• ✅ 支持批量操作

---

🎯 最终推荐方案

方案A: 混合多源（推荐）⭐⭐⭐⭐⭐

架构设计:

┌─────────────────────────────────────────────────────────┐
│  数据获取层                                             │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  实时行情:  新浪财经 (主) → 腾讯财经 (备)               │
│             └─ 响应时间: 30-50ms                        │
│             └─ 字段: 32个 / 40+个                       │
│                                                         │
│  历史K线:   腾讯财经 (主) → AkShare (备)                │
│             └─ 数据量: 320条 / 无限                     │
│             └─ 支持复权                                 │
│                                                         │
│  基本面:    东方财富 (主) → AkShare (补充)              │
│             └─ PE/PB 实时                               │
│             └─ 财务数据季度                             │
│                                                         │
├─────────────────────────────────────────────────────────┤
│  缓存层 (Redis)                                         │
├─────────────────────────────────────────────────────────┤
│  - 实时行情: TTL 60s                                    │
│  - 历史K线: TTL 3600s                                   │
│  - 基本面: TTL 86400s                                   │
└─────────────────────────────────────────────────────────┘

核心代码:

class MultiSourceDataFetcher:
    """多数据源获取器"""

    async def get_realtime_quote(self, symbol: str) -> Dict:
        """实时行情 - 自动容错"""

        # 检查缓存
        cache_key = f"quote:{symbol}"
        cached = await redis.get(cache_key)
        if cached:
            return cached

        # 尝试新浪
        try:
            quote = await self._fetch_sina(symbol)
            await redis.setex(cache_key, 60, quote)
            return quote
        except Exception as e:
            logger.warning(f"Sina failed: {e}")

        # 尝试腾讯
        try:
            quote = await self._fetch_tencent(symbol)
            await redis.setex(cache_key, 60, quote)
            return quote
        except Exception as e:
            logger.error(f"All sources failed: {e}")
            raise

    async def get_historical_kline(self, symbol: str, days: int = 365) -> pd.DataFrame:
        """历史K线"""

        # 检查缓存
        cache_key = f"kline:{symbol}:{days}"
        cached = await redis.get(cache_key)
        if cached:
            return pd.DataFrame(cached)

        # 优先腾讯（快速获取近期数据）
        if days <= 320:
            try:
                df = await self._fetch_tencent_kline(symbol, days)
                await redis.setex(cache_key, 3600, df.to_dict())
                return df
            except Exception as e:
                logger.warning(f"Tencent kline failed: {e}")

        # AkShare（完整历史数据）
        import akshare as ak
        df = ak.stock_zh_a_hist(symbol=symbol, period="daily", adjust="qfq")
        df_filtered = df.tail(days)
        await redis.setex(cache_key, 3600, df_filtered.to_dict())
        return df_filtered

    async def get_fundamentals(self, symbol: str) -> Dict:
        """基本面数据"""

        # 检查缓存
        cache_key = f"fund:{symbol}"
        cached = await redis.get(cache_key)
        if cached:
            return cached

        # 东方财富 - 实时PE/PB
        try:
            fund = await self._fetch_eastmoney_fundamentals(symbol)
            await redis.setex(cache_key, 86400, fund)
            return fund
        except Exception as e:
            logger.error(f"Fundamentals fetch failed: {e}")
            return {}

优点:

• ✅ 高可用（多数据源容错）
• ✅ 高性能（缓存 + 快速数据源）
• ✅ 数据全面（实时 + 历史 + 基本面）
• ✅ 完全免费
• ✅ 易于维护

缺点:

• ⚠️ 复杂度略高
• ⚠️ 需要维护多个数据源解析器

方案B: AkShare单一源（简化版）⭐⭐⭐⭐

适用场景: 开发阶段、小规模应用

import akshare as ak

# 所有数据通过AkShare获取
df_realtime = ak.stock_zh_a_spot_em()
df_kline = ak.stock_zh_a_hist(symbol="600000")
df_info = ak.stock_individual_info_em(symbol="600000")

优点:

• ✅ 极简单
• ✅ 接口统一
• ✅ 免费

缺点:

• ⚠️ 单点依赖
• ⚠️ 响应较慢

---

📋 API接口速查表

新浪财经

功能	URL	参数	示例
实时行情	http://hq.sinajs.cn/list=	股票代码（逗号分隔）	sh600000,sz000001
历史K线	http://money.finance.sina.com.cn/quotes_service/api/json_v2.php/CN_MarketData.getKLineData	symbol, scale, datalen	symbol=sh600000&scale=240&datalen=1023

腾讯财经

功能	URL	参数	示例
实时行情	http://qt.gtimg.cn/q=	股票代码（逗号分隔）	sh600000,sz000001
历史K线	http://web.ifzq.gtimg.cn/appstock/app/fqkline/get	param	param=sh600000,day,,,320,qfq

东方财富

功能	URL	参数	示例
实时行情	http://push2.eastmoney.com/api/qt/stock/get	secid, fields	secid=1.600000&fields=f43,f44,f45,f46
基本面	同上	secid, fields	secid=1.600000&fields=f162,f167

AkShare (Python库)

# 安装
pip install akshare

# 使用
import akshare as ak

# 实时行情
df = ak.stock_zh_a_spot_em()

# 历史K线
df = ak.stock_zh_a_hist(symbol="600000", period="daily", adjust="qfq")

# 个股信息
df = ak.stock_individual_info_em(symbol="600000")

---

🔧 实用工具函数

1. 统一数据格式转换

def normalize_quote(source: str, raw_data: Any) -> Dict:
    """统一不同数据源的行情格式"""

    if source == 'sina':
        values = raw_data.split(',')
        return {
            'symbol': extract_symbol(raw_data),
            'name': values[0],
            'price': float(values[3]),
            'open': float(values[1]),
            'high': float(values[4]),
            'low': float(values[5]),
            'volume': int(float(values[8])),
            'amount': float(values[9]),
            'source': 'sina'
        }

    elif source == 'tencent':
        values = raw_data.split('~')
        return {
            'symbol': values[2],
            'name': values[1],
            'price': float(values[3]),
            'open': float(values[5]),
            'high': float(values[33]),
            'low': float(values[34]),
            'volume': int(values[6]) * 100,  # 手转股
            'amount': float(values[37]),
            'source': 'tencent'
        }

    elif source == 'eastmoney':
        return {
            'symbol': raw_data['f57'],
            'name': raw_data['f58'],
            'price': raw_data['f43'] / 100,
            'open': raw_data['f46'] / 100,
            'high': raw_data['f44'] / 100,
            'low': raw_data['f45'] / 100,
            'volume': raw_data['f47'] * 100,  # 手转股
            'amount': raw_data['f48'],
            'turnover_rate': raw_data.get('f168', 0) / 100,
            'source': 'eastmoney'
        }

2. 限流控制

import asyncio
from functools import wraps

class RateLimiter:
    """API限流器"""

    def __init__(self, calls_per_second: float = 2):
        self.min_interval = 1.0 / calls_per_second
        self.last_called = 0.0

    async def acquire(self):
        """获取许可"""
        elapsed = time.time() - self.last_called
        if elapsed < self.min_interval:
            await asyncio.sleep(self.min_interval - elapsed)
        self.last_called = time.time()

# 使用示例
limiter = RateLimiter(calls_per_second=2)

async def fetch_with_limit(url):
    await limiter.acquire()
    return await fetch(url)

3. User-Agent 轮换

import random

USER_AGENTS = [
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36',
    'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36',
]

def get_headers(referer: str = None) -> Dict:
    """获取随机请求头"""
    headers = {
        'User-Agent': random.choice(USER_AGENTS),
        'Accept': 'text/html,application/json',
        'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
    }

    if referer:
        headers['Referer'] = referer

    return headers

---

📝 总结

✅ 最佳实践

1. 实时行情: 新浪财经（主） + 腾讯财经（备）
2. 历史K线: 腾讯财经（近期） + AkShare（长期）
3. 基本面: 东方财富 + AkShare
4. 缓存策略: Redis多层缓存（60s / 1h / 24h）
5. 容错机制: 多数据源自动切换
6. 限流控制: 2 QPS，避免被封

⚠️ 注意事项

1. 编码问题: 新浪、腾讯返回GBK编码，需要正确处理
2. 价格单位: 东方财富价格需除以100
3. 股票代码: 注意不同数据源的代码格式（sh600000 vs 1.600000）
4. JSONP解析: 腾讯、网易返回JSONP格式，需要去掉前缀
5. 反爬措施: 添加User-Agent、Referer、限流

🚀 下一步行动

1. 实现 backend/app/modules/data_fetcher_v2.py
2. 创建数据源适配器（新浪、腾讯、东方财富）
3. 实现缓存层和容错逻辑
4. 编写单元测试
5. 更新API接口，移除TuShare依赖

---

生成时间: 2025-12-21 测试版本: v1.0 建议复测周期: 每月