106 lines
3.3 KiB
Python
106 lines
3.3 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
财务数据采集模块
|
|
双轨策略：优先尝试 AKShare 在线采集，失败则回退到预置数据
|
|
"""
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
DATA_DIR = Path(__file__).parent.parent / "data"
|
|
|
|
|
|
def _load_preset_data() -> list:
|
|
"""加载预置的科创板企业数据"""
|
|
filepath = DATA_DIR / "sample_companies.json"
|
|
with open(filepath, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
|
|
|
|
def collect_financial_data(stock_code: str) -> Optional[dict]:
|
|
"""
|
|
采集指定股票代码的财务数据
|
|
双轨策略：在线采集 → 离线预置
|
|
"""
|
|
# 尝试在线采集
|
|
try:
|
|
return _collect_online(stock_code)
|
|
except Exception as e:
|
|
logger.warning(f"在线采集 {stock_code} 失败: {e}，回退到预置数据")
|
|
|
|
# 回退到预置数据
|
|
return _collect_from_preset(stock_code)
|
|
|
|
|
|
def _collect_online(stock_code: str) -> Optional[dict]:
|
|
"""通过 AKShare 在线采集财务数据"""
|
|
try:
|
|
import akshare as ak
|
|
|
|
# 科创板企业利润表
|
|
profit_df = ak.stock_profit_sheet_by_report_em(symbol=stock_code)
|
|
# 科创板企业资产负债表
|
|
balance_df = ak.stock_balance_sheet_by_report_em(symbol=stock_code)
|
|
|
|
if profit_df is not None and not profit_df.empty:
|
|
latest = profit_df.iloc[0]
|
|
return {
|
|
"stock_code": stock_code,
|
|
"revenue": float(latest.get("营业收入", 0)),
|
|
"net_profit": float(latest.get("净利润", 0)),
|
|
"rd_expense": float(latest.get("研发费用", 0)),
|
|
"source": "akshare_online",
|
|
}
|
|
except ImportError:
|
|
logger.warning("AKShare 未安装，跳过在线采集")
|
|
except Exception as e:
|
|
logger.warning(f"AKShare 采集异常: {e}")
|
|
|
|
raise RuntimeError("在线采集失败")
|
|
|
|
|
|
def _collect_from_preset(stock_code: str) -> Optional[dict]:
|
|
"""从预置数据中查找企业"""
|
|
companies = _load_preset_data()
|
|
for company in companies:
|
|
if company["stock_code"] == stock_code:
|
|
return {
|
|
"stock_code": stock_code,
|
|
"company_name": company["company_name"],
|
|
"industry": company["industry"],
|
|
"sector": company["sector"],
|
|
"financials": company["financials"],
|
|
"core_tech_personnel": company["core_tech_personnel"],
|
|
"tech_route": company["tech_route"],
|
|
"compliance": company["compliance"],
|
|
"supply_chain": company["supply_chain"],
|
|
"source": "preset_data",
|
|
}
|
|
return None
|
|
|
|
|
|
def get_all_companies() -> list:
|
|
"""获取所有预置企业列表"""
|
|
return _load_preset_data()
|
|
|
|
|
|
def get_company_by_code(stock_code: str) -> Optional[dict]:
|
|
"""通过股票代码查找企业完整数据"""
|
|
companies = _load_preset_data()
|
|
for company in companies:
|
|
if company["stock_code"] == stock_code:
|
|
return company
|
|
return None
|
|
|
|
|
|
def get_company_by_name(name: str) -> Optional[dict]:
|
|
"""通过企业名称查找（支持简称）"""
|
|
companies = _load_preset_data()
|
|
for company in companies:
|
|
if name in company["company_name"] or name in company["short_name"]:
|
|
return company
|
|
return None
|