Stock Research System — 股票研究系统

股票深度研究系统 专业级股权研究自动化系统，集成DataHub统一数据获取层 核心改进（2026-05-19） 集成DataHub统一数据获取层 替代多数据源直接调用，统一入口管理 智能缓存（quote: 5分钟, financial: 1小时） 自动降级机制（NeoData → WeStock → AkShare） 发布-订阅架构支持实时数据推送 简化数据获取流程 便捷函数：quick_quote(), quick_financial(), quick_macro() 生产管理器：get_production() 自然语言查询：quick_query() 增强代码示例 完整可执行的Python代码 明确的路径和依赖 统一的异常处理

执行流程 Step 1: 任务解析 提取参数： stock_code（必需）- 股票代码 company_name（必需）- 公司名称 report_type（默认 initiating_coverage）- 报告类型 market（自动推断）- 市场（A股/港股/美股） report_period（季报必需）- 报告期 缺少必需参数 → 提示用户补充

Step 2: 数据获取（data_fetch）- DataHub统一接口 数据源优先级（基于DataHub）： DataHub自动路由（优先级由Producer定义）：

• LocalWikiProducer（优先级1）- 本地Wiki知识库
• NeoDataProducer（优先级12）- 自然语言查询
• WeStockProducer（优先级13）- 腾讯自选股
• AkShareProducer（优先级15）- AkShare开源数据
• YFinanceProducer（优先级16）- Yahoo Finance

DataHub便捷函数使用：

import asyncio 
import sys 
# 用户需自行配置DataHub路径 
# sys.path.insert(0, "") 
from datahub.production_startup import ( 
    quick_quote, 
    quick_financial, 
    quick_macro, 
    quick_query, 
    get_production 
) 
async def fetch_research_data(stock_code: str, market: str = 'cn') -> dict: 
    """获取研究所需全部数据""" 
    prod = await get_production() 
    # 并发获取多维度数据 
    tasks = [ 
        quick_quote(stock_code, market=market), 
        quick_financial(stock_code, report_type='balance'), 
        quick_financial(stock_code, report_type='income'), 
        quick_financial(stock_code, report_type='cashflow'), 
    ] 
    results = await asyncio.gather(tasks, return_exceptions=True) 
    return { 
        'quote': results[0] if not isinstance(results[0], Exception) else None, 
        'balance': results[1] if not isinstance(results[1], Exception) else None, 
        'income': results[2] if not isinstance(results[2], Exception) else None, 
        'cashflow': results[3] if not isinstance(results[3], Exception) else None, 
    } 
# 执行 
asyncio.run(fetch_research_data("600519", market='cn')) 
自然语言查询示例： 
from datahub.production_startup import quick_query 
async def query_financial_data(query: str) -> dict: 
    """自然语言查询金融数据""" 
    result = await quick_query(query) 
    if result['status'] == 'success': 
        return result['data'] 
    else: 
        print(f"查询失败: {result.get('error')}") 
        return None 
# 示例查询 
asyncio.run(query_financial_data("贵州茅台最新财报营收和净利润")) 
asyncio.run(query_financial_data("新能源板块资金流向")) 
asyncio.run(query_financial_data("比亚迪机构评级和目标价")) 
主题订阅（高级用法）： 
import asyncio 
from datahub import get_datahub, register_all_producers 
async def subscribe_realtime_data(stock_code: str, callback): 
    """订阅实时数据推送""" 
    datahub = get_datahub() 
    register_all_producers(datahub) 
    topic = f"westock:cn:quote:{stock_code}" 
    datahub.subscribe(topic, callback) 
    await datahub.request_refresh(topic) 
# 示例回调 
async def on_quote_update(data): 
    print(f"行情更新: {data}") 
asyncio.run(subscribe_realtime_data("600519", on_quote_update)) 
货币单位注意： 
港股财报返回港元/美元，美股返回美元 
展示时必须标注正确货币单位，禁止使用人民币符号 
检查点： 
数据质量确认 - 是否继续？[Y/n] 
Step 3: 财务分析（financial_analysis） 
使用DataHub财务数据： 
from datahub.production_startup import quick_financial 
async def analyze_financials(stock_code: str) -> dict: 
    """财务分析""" 
    # 获取三大报表 
    balance = await quick_financial(stock_code, report_type='balance') 
    income = await quick_financial(stock_code, report_type='income') 
    cashflow = await quick_financial(stock_code, report_type='cashflow') 
    # 提取关键指标 
    if all(r['status'] == 'success' for r in [balance, income, cashflow]): 
        bal_data = balance['data'] 
        inc_data = income['data'] 
        cf_data = cashflow['data'] 
        # 计算核心指标 
        metrics = { 
            '毛利率': (inc_data['营收'] - inc_data['营业成本']) / inc_data['营收'], 
            '净利率': inc_data['净利润'] / inc_data['营收'], 
            'ROE': inc_data['净利润'] / bal_data['股东权益合计'], 
            'ROIC': calculate_roic(inc_data, bal_data), 
        } 
        return metrics 
    return None 
def calculate_roic(income: dict, balance: dict) -> float: 
    """计算ROIC""" 
    nopat = income['净利润'] + income['利息支出']  (1 - 0.25)  # 假设税率25% 
    invested_capital = balance['股东权益合计'] + balance['有息负债'] - balance['现金及等价物'] 
    return nopat / invested_capital if invested_capital > 0 else None 
异常处理： 
异常 处理 
指标无法计算 标注"N/A"，不估算 
负权益公司 用经营资产法计算 ROIC 
数据不足（<3年） 标注"分析可靠性降低" 
DataHub返回错误 检查Producer注册，切换备用源 
Step 4: 估值分析（valuation） 
DCF估值（Python计算）： 

python def dcf_valuation( free_cash_flow: float, growth_rate: float = 0.05, discount_rate: float = 0.10, terminal_growth: float = 0.03, years: int = 10 ) -> dict: """DCF估值计算""" # 预测未来现金流 fcf_projections = [] fcf = free_cash_flow for year in range(years): fcf = (1 + growth_rate) fcf_projections.append(fcf) # 折现 discounted_fcf = [] for i, fcf in enumerate(fcf_projections, 1): discounted = fcf / ((1 + discount_rate) i) discounted_fcf.append(discounted) # 终值 terminal_value = fcf_projections[-1] (1 + terminal_growth) / (discount_rate - terminal_growth) discounted_terminal = terminal_value / ((1 + discount_rate) ** years) ```