109 lines
3.7 KiB
Python
109 lines
3.7 KiB
Python
"""
|
|
Bloomberg 数据服务
|
|
|
|
负责处理 Bloomberg 特有的数据读取逻辑
|
|
从 stockcard 表中提取数据并转换为统一格式
|
|
"""
|
|
from typing import List, Dict, Optional
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import text
|
|
import logging
|
|
|
|
from app.models import Company
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
async def get_bloomberg_data(
|
|
company: Company,
|
|
db: AsyncSession
|
|
) -> List[Dict]:
|
|
"""
|
|
获取指定公司的 Bloomberg 财务数据
|
|
|
|
Args:
|
|
company: 公司对象
|
|
db: 数据库会话
|
|
|
|
Returns:
|
|
List[Dict]: 统一格式的财务数据列表
|
|
"""
|
|
try:
|
|
# 1. 查找对应的 Company_code
|
|
# stockcard 中存储的是 "AAPL US Equity" 而 symbol 是 "AAPL"
|
|
target_code = None
|
|
|
|
# 优先尝试最可能的后缀
|
|
suffixes = [" US Equity", " HK Equity", " JP Equity", " CH Equity", " VN Equity"]
|
|
possible_codes = [f"{company.symbol}{s}" for s in suffixes]
|
|
|
|
# 检查哪个存在
|
|
for code in possible_codes:
|
|
check_sql = text("SELECT 1 FROM stockcard WHERE Company_code = :code LIMIT 1")
|
|
exists = await db.execute(check_sql, {"code": code})
|
|
if exists.scalar():
|
|
target_code = code
|
|
break
|
|
|
|
# 如果没找到，尝试模糊匹配 (作为兜底)
|
|
if not target_code:
|
|
fuzzy_sql = text("SELECT Company_code FROM stockcard WHERE Company_code LIKE :symbol LIMIT 1")
|
|
fuzzy_res = await db.execute(fuzzy_sql, {"symbol": f"%{company.symbol}%"})
|
|
row = fuzzy_res.fetchone()
|
|
if row:
|
|
target_code = row[0]
|
|
|
|
if not target_code:
|
|
logger.warning(f"No Bloomberg data found for symbol: {company.symbol}")
|
|
return []
|
|
|
|
# 2. 获取该公司的所有数据
|
|
# schema: indicator, value, value_date (作为报告期)
|
|
query = text("""
|
|
SELECT indicator, value, value_date, currency
|
|
FROM stockcard
|
|
WHERE Company_code = :code
|
|
""")
|
|
result = await db.execute(query, {"code": target_code})
|
|
|
|
# 3. 数据透视 (Pivot)
|
|
data_by_date = {}
|
|
|
|
for row in result:
|
|
indicator = row.indicator
|
|
val = row.value
|
|
v_date = row.value_date
|
|
curr = row.currency
|
|
|
|
if not v_date:
|
|
continue
|
|
|
|
date_str = v_date.isoformat() if hasattr(v_date, 'isoformat') else str(v_date)
|
|
|
|
if date_str not in data_by_date:
|
|
data_by_date[date_str] = {
|
|
"end_date": date_str,
|
|
"currency": curr
|
|
}
|
|
elif data_by_date[date_str].get("currency") is None and curr:
|
|
# If existing entry has no currency, but we have one now, update it
|
|
data_by_date[date_str]["currency"] = curr
|
|
|
|
# Normalize key to match frontend expectation (mostly lowercase)
|
|
# e.g. "ROE" -> "roe", "PE" -> "pe"
|
|
norm_indicator = indicator.lower()
|
|
|
|
# Special case mapping if needed, but lowercase covers most
|
|
# frontend uses: net_income, revenue, etc.
|
|
|
|
data_by_date[date_str][norm_indicator] = val
|
|
|
|
# 4. 转换为列表并排序
|
|
full_list = list(data_by_date.values())
|
|
full_list.sort(key=lambda x: x['end_date'], reverse=True)
|
|
|
|
return full_list
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error fetching Bloomberg data from stockcard: {e}", exc_info=True)
|
|
return []
|