Finance Data Toolkit

金融数据抓取与分析全链路工具箱，采用分层架构：核心抓取框架常驻内存，细分功能模块按需加载。

核心架构（必读，覆盖 80% 场景）

1. 数据源分层

2. 抓取技术栈选型决策树

需要登录/复杂交互/动态渲染？
  ├─ 是 → browser-cdp (CDP控制真实 Chrome/Edge)
  │       ├─ 观察模式：用户操作，agent 建议/记录
  │       ├─ 代劳模式：agent 全自动执行
  │       └─ 协作模式：人机共同操作
  └─ 否 → requests/httpx + 解析器 (lxml/regex/json)
         ├─ 结构化 API → 直接调用
         ├─ 网页解析 → CSS/XPath + 正则
         └─ WebSocket/长连接 → aiohttp/websockets

3. 统一数据契约（所有模块输出标准化为此格式）

from dataclasses import dataclass
from datetime import datetime
from typing import Optional, Dict, Any

@dataclass
class FinanceData:
    source: str                    # 数据源标识: eastmoney/sina/ths/xueqiu/akshare...
    data_type: str                 # 数据类型: quote/kline/financial/news/guba/report
    symbol: str                    # 标的代码: 000001.SZ / 600000.SH / BTC-USDT
    timestamp: datetime            # 数据时间戳（统一 UTC）
    payload: Dict[str, Any]        # 业务载荷（见各模块 schema）
    raw: Optional[Dict] = None     # 原始响应（调试用，生产可关闭）
    meta: Optional[Dict] = None    # 元信息：请求耗时、重试次数、代理IP、版本号等

4. 核心抽象接口（所有抓取器实现此协议）

from abc import ABC, abstractmethod
from typing import AsyncIterator, List, Optional

class BaseScraper(ABC):
    """统一抓取器基类"""
    
    @property
    @abstractmethod
    def source_name(self) -> str: ...
    
    @property
    @abstractmethod
    def supported_types(self) -> List[str]: ...
    
    @abstractmethod
    async def fetch(self, 
        symbols: List[str],
        data_type: str,
        start: Optional[datetime] = None,
        end: Optional[datetime] = None,
        **kwargs
    ) -> AsyncIterator[FinanceData]: ...
    
    @abstractmethod
    async def health_check(self) -> bool: ...
    
    async def __aenter__(self): return self
    async def __aexit__(self, *args): await self.close()
    @abstractmethod
    async def close(self): ...

5. 并发与限流标准配置

# 推荐默认配置（可通过环境变量覆盖）
CONCURRENCY_LIMIT = 5           # 同一数据源并发请求数
REQUEST_TIMEOUT = 30            # 单请求超时(秒)
RETRY_TIMES = 3                 # 失败重试次数
RETRY_BACKOFF = [1, 2, 5]       # 指数退避(秒)
RATE_LIMIT_PER_MIN = 60         # 每分钟最大请求数
PROXY_ROTATION = True           # 是否轮换代理
BROWSER_POOL_SIZE = 2           # CDP 浏览器实例池大小

6. 增量更新策略

模块索引（按需加载）

Python 包结构

finance_toolkit/
├── __init__.py              # 统一入口，导出所有公开 API
├── core.py                  # 核心抽象：FinanceData, BaseScraper, create_scraper
├── exceptions.py            # 异常定义：FinanceError, SourceError, DataError 等
├── resilience.py            # 容错机制：CircuitBreaker, FallbackManager, retry_with_backoff
├── validation.py            # 数据验证：QuoteValidator, FinancialValidator, NewsValidator 等
├── scrapers/                # 具体数据源抓取器
│   ├── __init__.py
│   ├── akshare_scraper.py   # AKShare 抓取器
│   ├── tushare_scraper.py   # Tushare 抓取器
│   ├── eastmoney_scraper.py # 东方财富抓取器
│   ├── sina_scraper.py      # 新浪财经抓取器
│   └── yahoo_scraper.py     # Yahoo Finance 抓取器
├── data_fetching/           # 数据获取高级封装
│   ├── __init__.py
│   ├── fetchers.py          # 同步/异步获取函数
│   ├── async_fetchers.py    # 异步获取函数
│   ├── eastmoney_fetcher.py # 东方财富实时行情/财务/资金流向
│   ├── sina_kline_fetcher.py # 新浪 K 线 + 技术指标
│   └── guba_scraper.py      # 东方财富股吧帖子/评论/用户画像抓取
├── technical_analysis/      # 技术指标计算与信号生成
│   ├── __init__.py
│   └── indicators.py        # MA/MACD/RSI/BOLL/KDJ 等
├── backtesting/             # 因子回测框架
│   ├── __init__.py
│   └── backtest_framework.py
├── sentiment/               # 舆情分析
│   ├── __init__.py
│   └── sentiment_analyzer.py
├── batch_processing/        # 批量数据处理
│   ├── __init__.py
│   └── batch_fetcher.py
├── report_generation/       # 研报生成
│   ├── __init__.py
│   └── report_generator.py
├── cleaning/                # 数据清洗标准化
│   ├── __init__.py
│   ├── alignment.py         # 时间对齐
│   ├── dedup.py             # 去重
│   ├── features.py          # 特征工程
│   ├── mappers.py           # 字段映射
│   ├── missing.py           # 缺失值处理
│   ├── normalizers.py       # 标准化
│   ├── pipeline.py          # 清洗流水线
│   ├── quality.py           # 质量监控
│   └── validators.py        # 数据验证
├── news/                    # 财经新闻抓取
│   ├── __init__.py
│   ├── models.py            # 新闻数据模型
│   ├── scrapers.py          # 多源新闻抓取器
│   ├── aggregator.py        # 新闻聚合
│   └── stream.py            # 实时新闻流
├── social/                  # 社交媒体数据
│   ├── __init__.py
│   ├── models.py            # 社交数据模型
│   └── scrapers.py          # 社交媒体抓取器

快速开始示例

场景 1：获取某股票完整基础数据

from finance_toolkit import create_scraper

async def get_stock_full_data(symbol: str):
    async with create_scraper('akshare') as scraper:
        async for data in scraper.fetch([symbol], 'quote'):
            print(f"实时行情: {data.payload}")
        async for data in scraper.fetch([symbol], 'kline', period='daily'):
            print(f"日K线: {len(data.payload)} 条")
        async for data in scraper.fetch([symbol], 'financial'):
            print(f"财务报表: {data.payload.keys()}")

场景 2：一键分析股票（获取K线+计算指标+生成信号）

from finance_toolkit import analyze_stock

result = analyze_stock('600000.SH')
print(f"当前价: {result['price_stats']['current_price']}")
print(f"技术评分: {result['latest_indicators']}")
print(f"交易信号: {result['signals']}")

场景 3：运行因子回测

from finance_toolkit import run_backtest

backtest_result = run_backtest(
    symbols=['600000.SH', '000001.SZ', '600519.SH'],
    periods=5,
    n_long=3,
    n_short=3,
    rebalance='W',
    method='equal_weight',
    long_only=False
)
print(f"年化收益: {backtest_result['annualized_return']:.2%}")
print(f"夏普比率: {backtest_result['sharpe_ratio']:.4f}")
print(f"最大回撤: {backtest_result['max_drawdown']:.2%}")

场景 4：舆情分析

from finance_toolkit import analyze_sentiment, analyze_stock_sentiment

# 单条文本
result = analyze_sentiment("茅台今天大涨5%，业绩超预期，机构纷纷买入，北向资金大幅流入！")
print(f"情感: {result.payload['label']} (score={result.payload['score']:.3f})")

# 股票舆情聚合
posts = [
    {'content': '茅台业绩超预期，看好后市', 'read_count': 10000, 'comment_count': 500},
    {'content': '茅台估值太高了，泡沫风险大', 'read_count': 5000, 'comment_count': 200},
]
agg = analyze_stock_sentiment(posts, symbol='600519.SH')
print(f"舆情信号: {agg.payload['signal']['signal']}")

场景 5：生成综合研报

from finance_toolkit import generate_comprehensive_report

files = generate_comprehensive_report(
    codes=['600000.SH', '000001.SZ', '600519.SH'],
    formats=['html', 'md', 'json']
)
print(f"报告已生成: {files}")

场景 6：批量获取数据

from finance_toolkit import batch_fetch_stocks, batch_fetch_klines

# 批量获取实时行情
summary = batch_fetch_stocks(['600000.SH', '000001.SZ', '600519.SH'])
print(f"成功: {summary['success']}, 失败: {summary['failed']}")

# 批量获取K线（需先启动 browser-cdp）
klines = batch_fetch_klines(['600000.SH', '000001.SZ'], port=9333, tab_id='xxx')

场景 7：东方财富股吧舆情抓取

from finance_toolkit import (
    sync_fetch_guba_posts,
    sync_fetch_guba_hot_posts,
    GubaPost,
    GubaComment,
    GubaUserProfile,
    EastmoneyGubaAPI,
    GubaCDPScraper,
)

# 同步获取某股票最新帖子
posts = sync_fetch_guba_posts('600519', page=1, page_size=10, sort='time')
for p in posts:
    print(f"{p.post_id}: {p.title[:30]}... (阅读:{p.read_count}, 评论:{p.comment_count})")

# 获取概念板块热帖
hot_posts = sync_fetch_guba_hot_posts('concept', top_n=5)
for p in hot_posts:
    print(f"{p.post_id}: {p.title[:30]}... (阅读:{p.read_count})")

# 异步用法（需配合 CDP 浏览器）
# async with EastmoneyGubaAPI() as api:
#     async for post in api.get_post_list(stock_code='600519', page=1, sort='hot'):
#         print(post.title)
# 
# async with GubaCDPScraper() as scraper:
#     detail = await scraper.get_post_detail('https://guba.eastmoney.com/news,123456.html')
#     comments = await scraper.get_comment_tree('123456')
#     profile = await scraper.get_user_profile('user123')

数据验证与合规检查

合规检查器

from finance_toolkit.compliance_checker import (
    ComplianceChecker,
    check_compliance,
    check_compliance_batch,
    get_compliance_summary,
)

checker = ComplianceChecker()
result = checker.check(data_dict)

# 批量检查
results = checker.check_batch(data_list)
summary = checker.get_summary(results)

数据质量验证

from finance_toolkit.validation import (
    validate_kline_data,
    validate_quote_data,
    DataQualityValidator,
    check_data_quality,
)

# 验证 K 线数据
report = validate_kline_data(df)

# 验证实时行情
report = validate_quote_data(quote, symbol='600000.SH')

# 自动检测类型
report = check_data_quality(data)

输出格式化

from finance_toolkit.output_formatter import (
    ComplianceOutputFormatter,
    format_compliance_result,
)

formatter = ComplianceOutputFormatter()
json_output = formatter.to_json(result)
text_output = formatter.to_text(result)
table_output = formatter.to_table(result)
summary_output = formatter.to_summary(result)

示例脚本

运行示例脚本验证合规检查功能：

# 数据质量校验示例
cd .claude/skills/finance-data-toolkit
python examples/data_validation_example.py

# 合规检查示例
python examples/compliance_check_example.py

# 数据验证脚本（支持多种输出格式）
python examples/data_validation_script.py --format json
python examples/data_validation_script.py --format text
python examples/data_validation_script.py --format table
python examples/data_validation_script.py --format summary
python examples/data_validation_script.py --output-dir ./output

索引说明

• 主文件只保留架构决策、统一契约、快速入口，不包含具体实现细节
• 具体抓取逻辑、参数详解、代码示例请加载对应 resources 子资源
• 超大型参考资料（完整 API 手册、历史数据字典、示例 Notebook）走 browse_paths，请自行 grep/view 检索
• 所有子资源文件均在 references/ 目录下，已在 frontmatter resources/browse_paths 登记
• 示例脚本位于 examples/ 目录，详见 examples/README.md

环境变量配置

# 必需：数据源 Token
AKSHARE_TOKEN=your_token
TUSHARE_TOKEN=your_token
WIND_USERNAME=xxx
WIND_PASSWORD=xxx

# 可选：代理/浏览器/并发
HTTP_PROXY=http://127.0.0.1:7890
HTTPS_PROXY=http://127.0.0.1:7890
CDP_ENDPOINT=http://127.0.0.1:9222  # 已打开的 Chrome CDP 端口
CONCURRENCY_LIMIT=5
RATE_LIMIT_PER_MIN=60
LOG_LEVEL=INFO

依赖建议

# 核心
akshare>=1.12.0
tushare>=1.2.80
pandas>=2.0
numpy>=1.24
pydantic>=2.5
httpx>=0.25
lxml>=4.9

# 浏览器自动化
browser-cdp>=0.3.0  # 本项目内置 skill
playwright>=1.40  # 备选

# 分析/可视化
talib-binary>=0.4.24  # 技术指标
matplotlib>=3.7
plotly>=5.17
jinja2>=3.1  # 模板引擎

# NLP
jieba>=0.42
snownlp>=0.12.3
transformers>=4.36  # 可选：深度学习模型

维护提示：新增数据源/模块时，请同步更新：

1. 本文件的「模块索引」表
2. 对应的 references/*.md 子资源
3. references/full-api-docs/ 下的 API 手册
4. references/example-notebooks/ 下的演示 Notebook

迁移记录：2026-07-15 将 browser-cdp 中的金融相关脚本迁移至 finance_toolkit 包：

• fetch_eastmoney_stock.py → finance_toolkit.data_fetching.eastmoney_fetcher
• fetch_kline_sina.py → finance_toolkit.data_fetching.sina_kline_fetcher
• calc_technical_indicators.py → finance_toolkit.technical_analysis.indicators
• backtest_framework.py → finance_toolkit.backtesting.backtest_framework
• sentiment_analysis.py → finance_toolkit.sentiment.sentiment_analyzer
• batch_fetch_stocks.py / batch_fetch_kline.py → finance_toolkit.batch_processing.batch_fetcher
• generate_report.py / generate_comprehensive_report.py → finance_toolkit.report_generation.report_generator
• 新增统一入口 finance_toolkit/init.py 提供 create_scraper, analyze_stock, run_backtest 等便捷函数

目录重构：2026-08-02 将 browser-cdp skill 迁移至子目录模块化结构：

• core/ → cdp_client.py, utils.py
• browser_ops/ → browser_launch.py, browser_nav.py, browser_extract.py, browser_screenshot.py, browser_input.py, browser_console.py, browser_watch.py, browser_smart_wait.py, browser_infinite_scroll.py, browser_captcha.py
• search_engines/ → baidu_search.py, bing_search.py, zhihu_search.py, zhihu_hot.py, zhihu_column_search.py, zhihu_publish_answer.py, arxiv_search.py, arxiv_multi_search.py, wechat_search.py
• references/ → 所有子资源文档
• 根目录保留 demo/test 脚本和 SKILL.md

统一接口增强：2026-08-08 新增统一数据抓取接口与异常处理机制：

• finance_toolkit/exceptions.py → 增强异常体系（新增 RateLimitError, ConnectionError, TimeoutError, DataIntegrityError, SourceHealthError）
• finance_toolkit/resilience.py → 增强容错机制（CircuitBreaker, FallbackManager, RateLimiter, HealthChecker）
• finance_toolkit/unified_fetcher.py → 统一数据抓取接口（UnifiedFetcher）
• finance_toolkit/health_monitor.py → 健康监控模块（HealthMonitor, AlertHandler）
• finance_toolkit/rate_limiter.py → 限流器模块（TokenBucket, LeakyBucket, SlidingWindowCounter）
• tests/test_unified_interface.py → 33 个测试用例全部通过