feat: 通用市场财务/快照/分析接口;增强数据源与配置读取

Backend
- router(financial): 新增通用路径 /{market}/{stock_code}、/snapshot、/analysis/stream
  - 用 MarketEnum 统一市场(cn/us/hk/jp)
  - 将 /china/{ts_code} 改为通用 get_financials,并规范 period,按年限裁剪
  - 新增通用昨日快照接口(CN 复用原逻辑,其他市场兜底近交易日收盘)
- data_manager: 仅从 config/config.json 读取各 provider API key,不再读取环境变量
  - series 构建更健壮:None/空结构判定;接受 numpy/pandas 数值类型并安全转 float
- provider(finnhub):
  - SDK 失败时使用 httpx 直连兜底(profile2、financials-reported)
  - 规范化年度报表,映射 revenue/net income/gross profit/assets/equity/goodwill/OCF/CapEx
  - 计算 gross/net margin、ROA、ROE;直接产出 series 结构
  - 增加关键步骤日志与异常保护
- provider(yfinance): 修正同步阻塞的获取逻辑,使用 run_in_executor 包装

Frontend
- hooks(useApi):
  - 将中国财务接口路径改为 /api/financials/cn
  - 新增 useFinancials 与 useSnapshot,统一多市场数据访问
- report/[symbol]/page.tsx:
  - 支持多市场(映射 usa→us、china→cn 等),统一 symbol 与分析流路径
  - 去除仅限中国市场的 UI 限制,财务/分析/图表对多市场可用
  - 使用新的分析与快照 API 路径
- lib/prisma.ts: 去除无关内容(微小空行调整)

Docs
- 重组文档目录:
  - docs/已完成任务/tasks.md(重命名自 docs/tasks.md)
  - docs/未完成任务/us_market_integration_tasks.md 新增

BREAKING CHANGE
- API 路径变更:
  - 财务数据:/api/financials/china/{ts_code} → /api/financials/{market}/{stock_code}
  - 快照:/api/financials/china/{ts_code}/snapshot → /api/financials/{market}/{stock_code}/snapshot
  - 分析流:/api/financials/china/{ts_code}/analysis/{type}/stream → /api/financials/{market}/{stock_code}/analysis/{type}/stream
- 前端需使用 useFinancials/useSnapshot 或更新为 /cn 路径以兼容中国市场
This commit is contained in:
xucheng 2025-11-06 19:58:36 +08:00
parent 0b09abf2e5
commit ca60410966
10 changed files with 519 additions and 136 deletions

View File

@ -2,6 +2,7 @@ import yaml
import os
import json
from typing import Any, Dict, List, Optional
from numbers import Number
from app.data_providers.base import BaseDataProvider
from app.data_providers.tushare import TushareProvider
# from app.data_providers.ifind import TonghsProvider
@ -44,20 +45,9 @@ class DataManager:
self.providers = {}
# Build provider base config from environment variables and config/config.json, then initialize providers
# Build provider base config ONLY from config/config.json (do not read env vars)
base_cfg: Dict[str, Any] = {"data_sources": {}}
# 1) Prefer env vars when present
for name, source_config in (self.config.get('data_sources') or {}).items():
env_var = source_config.get('api_key_env')
if env_var:
api_key = os.getenv(env_var)
if api_key:
base_cfg["data_sources"][name] = {"api_key": api_key}
else:
logger.warning(f"Env var '{env_var}' for provider '{name}' not set; will try config.json.")
# 2) Fallback to config/config.json if tokens are provided there
try:
# Use the same REPO_ROOT calculation as data_sources.yaml
current_dir = os.path.dirname(__file__)
@ -76,7 +66,7 @@ class DataManager:
cfg_json = json.load(jf)
ds_from_json = (cfg_json.get("data_sources") or {})
for name, node in ds_from_json.items():
if name not in base_cfg["data_sources"] and node.get("api_key"):
if node.get("api_key"):
base_cfg["data_sources"][name] = {"api_key": node.get("api_key")}
logger.info(f"Loaded API key for provider '{name}' from config.json")
else:
@ -106,11 +96,9 @@ class DataManager:
}
for name, provider_class in provider_map.items():
token = None
token = base_cfg.get("data_sources", {}).get(name, {}).get("api_key")
source_config = self.config['data_sources'].get(name, {})
if source_config and source_config.get('api_key_env'):
token = base_cfg.get("data_sources", {}).get(name, {}).get("api_key")
# Initialize the provider if a token is found or not required
if token or not source_config.get('api_key_env'):
try:
@ -118,7 +106,7 @@ class DataManager:
except Exception as e:
logger.error(f"Failed to initialize provider '{name}': {e}")
else:
logger.warning(f"Provider '{name}' requires token env '{source_config.get('api_key_env')}', but none provided. Skipping.")
logger.warning(f"Provider '{name}' requires API key but none provided in config.json. Skipping.")
def _detect_market(self, stock_code: str) -> str:
if stock_code.endswith(('.SH', '.SZ')):
@ -143,7 +131,17 @@ class DataManager:
try:
method = getattr(provider, method_name)
data = await method(stock_code=stock_code, **kwargs)
if data is not None and (not isinstance(data, list) or data):
is_success = False
if data is None:
is_success = False
elif isinstance(data, list):
is_success = len(data) > 0
elif isinstance(data, dict):
is_success = len(data) > 0
else:
is_success = True
if is_success:
logger.info(f"Data successfully fetched from '{provider_name}' for '{stock_code}'.")
return data
except Exception as e:
@ -171,11 +169,18 @@ class DataManager:
for key, value in report.items():
if key in ['ts_code', 'stock_code', 'year', 'end_date', 'period', 'ann_date', 'f_ann_date', 'report_type']:
continue
if isinstance(value, (int, float)) and value is not None:
# Accept numpy/pandas numeric types as well as builtin numbers
if value is not None and isinstance(value, Number):
if key not in series:
series[key] = []
if not any(d['year'] == year for d in series[key]):
series[key].append({"year": year, "value": value})
# Store as builtin float to avoid JSON serialization issues
try:
numeric_value = float(value)
except Exception:
# Fallback: skip if cannot coerce to float
continue
series[key].append({"year": year, "value": numeric_value})
return series
else:
return {}

View File

@ -5,6 +5,7 @@ import pandas as pd
from datetime import datetime, timedelta
import asyncio
import logging
import httpx
logger = logging.getLogger(__name__)
@ -14,11 +15,37 @@ class FinnhubProvider(BaseDataProvider):
if not self.token:
raise ValueError("Finnhub API key not provided.")
self.client = finnhub.Client(api_key=self.token)
try:
masked = f"***{self.token[-4:]}" if isinstance(self.token, str) and len(self.token) >= 4 else "***"
logger.info(f"[Finnhub] client initialized (token={masked})")
except Exception:
# 避免日志失败影响初始化
pass
async def get_stock_basic(self, stock_code: str) -> Optional[Dict[str, Any]]:
async def _fetch():
def _fetch():
try:
profile = self.client.company_profile2(symbol=stock_code)
profile = None
try:
profile = self.client.company_profile2(symbol=stock_code)
logger.debug(f"[Finnhub] SDK company_profile2 ok symbol={stock_code} name={profile.get('name') if isinstance(profile, dict) else None}")
except Exception as e:
logger.warning(f"[Finnhub] SDK company_profile2 failed for {stock_code}: {e}")
# Fallback to direct HTTP if SDK call fails
try:
resp = httpx.get(
'https://finnhub.io/api/v1/stock/profile2',
params={'symbol': stock_code},
headers={'X-Finnhub-Token': self.token},
timeout=20.0,
)
logger.debug(f"[Finnhub] HTTP profile2 status={resp.status_code} len={len(resp.text)}")
if resp.status_code == 200:
profile = resp.json()
else:
logger.error(f"[Finnhub] HTTP profile2 failed status={resp.status_code} body={resp.text[:200]}")
except Exception:
profile = None
if not profile:
return None
@ -39,13 +66,18 @@ class FinnhubProvider(BaseDataProvider):
return await loop.run_in_executor(None, _fetch)
async def get_daily_price(self, stock_code: str, start_date: str, end_date: str) -> List[Dict[str, Any]]:
async def _fetch():
def _fetch():
try:
start_ts = int(datetime.strptime(start_date, '%Y%m%d').timestamp())
end_ts = int(datetime.strptime(end_date, '%Y%m%d').timestamp())
logger.debug(f"[Finnhub] stock_candles symbol={stock_code} D {start_date}->{end_date}")
res = self.client.stock_candles(stock_code, 'D', start_ts, end_ts)
if res.get('s') != 'ok':
try:
logger.warning(f"[Finnhub] stock_candles not ok symbol={stock_code} status={res.get('s')}")
except Exception:
pass
return []
df = pd.DataFrame(res)
@ -68,76 +100,211 @@ class FinnhubProvider(BaseDataProvider):
return await loop.run_in_executor(None, _fetch)
async def get_financial_statements(self, stock_code: str, report_dates: List[str]) -> List[Dict[str, Any]]:
async def _fetch():
def _fetch():
try:
# Finnhub provides financials as a whole, not by specific date ranges in one call
# We fetch all available and then filter.
# Note: 'freq' can be 'annual' or 'quarterly'. We'll use annual.
res = self.client.financials_reported(symbol=stock_code, freq='annual')
# 1) 拉取年度报表financials_reported, annual
res = None
try:
res = self.client.financials_reported(symbol=stock_code, freq='annual')
except Exception as e:
logger.warning(f"[Finnhub] SDK financials_reported failed for {stock_code}: {e}")
# Fallback: direct HTTP
try:
r = httpx.get(
'https://finnhub.io/api/v1/stock/financials-reported',
params={'symbol': stock_code, 'freq': 'annual'},
headers={'X-Finnhub-Token': self.token},
timeout=30.0,
)
logger.debug(f"[Finnhub] HTTP financials-reported status={r.status_code} len={len(r.text)}")
if r.status_code == 200:
res = r.json()
else:
logger.error(f"[Finnhub] HTTP financials-reported failed status={r.status_code} body={r.text[:300]}")
except Exception:
res = None
if not res or not res.get('data'):
logger.warning(f"[Finnhub] financials-reported empty for {stock_code}")
return []
df = pd.DataFrame(res['data'])
if df.empty:
logger.warning(f"[Finnhub] financials-reported dataframe empty for {stock_code}")
return []
# Filter by requested dates
years_to_fetch = {date[:4] for date in report_dates}
df = df[df['year'].astype(str).isin(years_to_fetch)]
# 2) 仅保留请求的年份
years_to_fetch = {str(date)[:4] for date in report_dates}
logger.debug(f"[Finnhub] filter years {sorted(list(years_to_fetch))} before={len(df)}")
if 'year' in df.columns:
df = df[df['year'].astype(str).isin(years_to_fetch)]
# 兜底:如果缺失 year 列,则用 endDate 推断
if 'year' not in df.columns and 'endDate' in df.columns:
df = df[df['endDate'].astype(str).str[:4].isin(years_to_fetch)]
# The data is deeply nested in 'report'. We need to extract and pivot it.
all_reports = []
for index, row in df.iterrows():
report_data = {'ts_code': stock_code, 'end_date': row['endDate']}
# Extract concepts from balance sheet, income statement, and cash flow
for item in row['report'].get('bs', []):
report_data[item['concept']] = item['value']
for item in row['report'].get('ic', []):
report_data[item['concept']] = item['value']
for item in row['report'].get('cf', []):
report_data[item['concept']] = item['value']
all_reports.append(report_data)
if df.empty:
logger.warning(f"[Finnhub] financials-reported no rows after filter for {stock_code}")
return []
# Further normalization of keys would be needed here to match a common format
# e.g. 'AssetsTotal' -> 'total_assets'
# This is a complex task and depends on the desired final schema.
# We will now normalize and calculate derived metrics
normalized_reports = []
for report in all_reports:
normalized_report = {
"ts_code": report.get("ts_code"),
"end_date": report.get("end_date"),
# Balance Sheet
"total_assets": report.get("AssetsTotal"),
"total_liabilities": report.get("LiabilitiesTotal"),
"equity": report.get("StockholdersEquityTotal"),
# Income Statement
"revenue": report.get("RevenuesTotal"),
"net_income": report.get("NetIncomeLoss"),
"gross_profit": report.get("GrossProfit"),
# Cash Flow
"net_cash_flow_operating": report.get("NetCashFlowOperating"),
def _normalize_key(s: Optional[str]) -> str:
if not isinstance(s, str):
return ""
return ''.join(ch.lower() for ch in s if ch.isalnum())
def pick(report_block: List[Dict[str, Any]], concept_candidates: List[str], label_candidates: List[str] = []) -> Optional[float]:
if not report_block:
return None
try:
by_concept = { _normalize_key(item.get('concept')): item.get('value') for item in report_block if isinstance(item, dict) }
by_label = { _normalize_key(item.get('label')): item.get('value') for item in report_block if isinstance(item, dict) }
except Exception:
return None
for key in concept_candidates:
v = by_concept.get(_normalize_key(key))
if v is not None:
try:
return float(v)
except Exception:
continue
for key in label_candidates:
v = by_label.get(_normalize_key(key))
if v is not None:
try:
return float(v)
except Exception:
continue
return None
# 3) 遍历年度记录,展开并标准化字段名
flat_reports: List[Dict[str, Any]] = []
for _, row in df.iterrows():
bs = (row.get('report') or {}).get('bs', [])
ic = (row.get('report') or {}).get('ic', [])
cf = (row.get('report') or {}).get('cf', [])
end_date = str(row.get('endDate') or '')
revenue = pick(
ic,
concept_candidates=['Revenues', 'RevenueFromContractWithCustomerExcludingAssessedTax', 'SalesRevenueNet', 'Revenue', 'RevenuesNet', 'SalesRevenueGoodsNet'],
label_candidates=['Total revenue', 'Revenue', 'Sales revenue']
)
net_income = pick(
ic,
concept_candidates=['NetIncomeLoss', 'ProfitLoss', 'NetIncomeLossAvailableToCommonStockholdersBasic', 'NetIncomeLossAvailableToCommonStockholdersDiluted'],
label_candidates=['Net income', 'Net income (loss)']
)
gross_profit = pick(
ic,
concept_candidates=['GrossProfit'],
label_candidates=['Gross profit']
)
total_assets = pick(
bs,
concept_candidates=['Assets', 'AssetsTotal', 'AssetsCurrentAndNoncurrent', 'AssetsIncludingAssetsMeasuredAtFairValue'],
label_candidates=['Total assets']
)
total_equity = pick(
bs,
concept_candidates=['StockholdersEquityIncludingPortionAttributableToNoncontrollingInterest', 'StockholdersEquity', 'StockholdersEquityTotal', 'Equity'],
label_candidates=['Total equity', "Stockholders' equity"]
)
goodwill = pick(
bs,
concept_candidates=['Goodwill', 'GoodwillAndIntangibleAssets'],
label_candidates=['Goodwill', 'Goodwill and intangible assets']
)
n_cashflow_act = pick(
cf,
concept_candidates=['NetCashProvidedByUsedInOperatingActivities', 'NetCashProvidedByUsedInOperatingActivitiesContinuingOperations', 'NetCashFlowOperating'],
label_candidates=['Net cash provided by operating activities']
)
capex = pick(
cf,
concept_candidates=['CapitalExpenditures', 'PaymentsToAcquirePropertyPlantAndEquipment', 'PaymentsToAcquireProductiveAssets'],
label_candidates=['Capital expenditures']
)
# 计算衍生指标
free_cash_flow = None
if isinstance(n_cashflow_act, (int, float)) and isinstance(capex, (int, float)):
free_cash_flow = n_cashflow_act - capex
normalized = {
# 基本元字段
'ts_code': stock_code,
'end_date': end_date, # DataManager 会从这里抽取 year
# 标准命名(见 financial_data_dictionary
'revenue': revenue,
'n_income': net_income,
'gross_profit': gross_profit,
'total_assets': total_assets,
'total_hldr_eqy_exc_min_int': total_equity,
'goodwill': goodwill,
'n_cashflow_act': n_cashflow_act,
'c_pay_acq_const_fiolta': capex,
'__free_cash_flow': free_cash_flow,
}
# Calculate derived metrics
if normalized_report["revenue"] and normalized_report["revenue"] > 0:
normalized_report["gross_margin"] = (normalized_report["gross_profit"] / normalized_report["revenue"]) if normalized_report["gross_profit"] else None
normalized_report["net_margin"] = (normalized_report["net_income"] / normalized_report["revenue"]) if normalized_report["net_income"] else None
if normalized_report["total_assets"] and normalized_report["total_assets"] > 0:
normalized_report["roa"] = (normalized_report["net_income"] / normalized_report["total_assets"]) if normalized_report["net_income"] else None
# 一些常用比率(若有足够数据则计算),命名对齐文档
if isinstance(revenue, (int, float)) and revenue > 0 and isinstance(gross_profit, (int, float)):
normalized['grossprofit_margin'] = gross_profit / revenue
if isinstance(revenue, (int, float)) and revenue > 0 and isinstance(net_income, (int, float)):
normalized['netprofit_margin'] = net_income / revenue
if isinstance(total_assets, (int, float)) and total_assets > 0 and isinstance(net_income, (int, float)):
normalized['roa'] = net_income / total_assets
if isinstance(total_equity, (int, float)) and total_equity > 0 and isinstance(net_income, (int, float)):
normalized['roe'] = net_income / total_equity
if normalized_report["equity"] and normalized_report["equity"] > 0:
normalized_report["roe"] = (normalized_report["net_income"] / normalized_report["equity"]) if normalized_report["net_income"] else None
normalized_reports.append(normalized_report)
flat_reports.append(normalized)
try:
logger.debug(
f"[Finnhub] row endDate={end_date} revenue={revenue} net_income={net_income} gross_profit={gross_profit} "
f"assets={total_assets} equity={total_equity} goodwill={goodwill} n_cfo={n_cashflow_act} capex={capex}"
)
except Exception:
pass
return normalized_reports
# Convert flat reports to series dict directly to match DataManager expected format
series: Dict[str, List[Dict[str, Any]]] = {}
for report in flat_reports:
end_date = str(report.get('end_date') or '')
year = end_date[:4] if len(end_date) >= 4 else None
if not year:
continue
period = f"{year}1231"
for key, value in report.items():
if key in ['ts_code', 'end_date']:
continue
# Only collect numeric values
try:
if value is None:
continue
num = float(value)
except Exception:
continue
if key not in series:
series[key] = []
# Avoid duplicate period entries
exists = any(dp.get('period') == period for dp in series[key])
if not exists:
series[key].append({'period': period, 'value': num})
try:
total_points = sum(len(v) for v in series.values())
logger.info(f"[Finnhub] built series for {stock_code} keys={len(series)} points={total_points}")
except Exception:
pass
return series
except Exception as e:
logger.error(f"Finnhub get_financial_statements failed for {stock_code}: {e}")
return []
loop = asyncio.get_event_loop()
return await loop.run_in_executor(None, _fetch)

View File

@ -21,7 +21,7 @@ class YfinanceProvider(BaseDataProvider):
return stock_code
async def get_stock_basic(self, stock_code: str) -> Optional[Dict[str, Any]]:
async def _fetch():
def _fetch():
try:
ticker = yf.Ticker(self._map_stock_code(stock_code))
info = ticker.info
@ -44,7 +44,7 @@ class YfinanceProvider(BaseDataProvider):
return await loop.run_in_executor(None, _fetch)
async def get_daily_price(self, stock_code: str, start_date: str, end_date: str) -> List[Dict[str, Any]]:
async def _fetch():
def _fetch():
try:
# yfinance date format is YYYY-MM-DD
start_fmt = datetime.strptime(start_date, '%Y%m%d').strftime('%Y-%m-%d')

View File

@ -5,6 +5,7 @@ import json
import os
import time
from datetime import datetime, timezone, timedelta
from enum import Enum
from typing import Dict, List
from fastapi import APIRouter, HTTPException, Query
@ -46,6 +47,13 @@ def get_dm():
router = APIRouter()
class MarketEnum(str, Enum):
cn = "cn"
us = "us"
hk = "hk"
jp = "jp"
# Load metric config from file (project root is repo root, not backend/)
# routers/ -> app/ -> backend/ -> repo root
REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", ".."))
@ -257,9 +265,10 @@ async def get_financial_config():
return FinancialConfigResponse(api_groups=api_groups)
@router.get("/china/{ts_code}", response_model=BatchFinancialDataResponse)
async def get_china_financials(
ts_code: str,
@router.get("/{market}/{stock_code}", response_model=BatchFinancialDataResponse)
async def get_financials(
market: MarketEnum,
stock_code: str,
years: int = Query(10, ge=1, le=10),
):
# Load metric config
@ -274,11 +283,11 @@ async def get_china_financials(
steps: List[StepRecord] = []
# Get company name
company_name = ts_code
company_name = stock_code
try:
basic_data = await get_dm().get_stock_basic(stock_code=ts_code)
basic_data = await get_dm().get_stock_basic(stock_code=stock_code)
if basic_data:
company_name = basic_data.get("name", ts_code)
company_name = basic_data.get("name", stock_code)
except Exception:
pass # Continue without it
@ -295,7 +304,7 @@ async def get_china_financials(
steps.append(step_financials)
# Fetch all financial statements at once (already in series format from provider)
series = await get_dm().get_financial_statements(stock_code=ts_code, report_dates=report_dates)
series = await get_dm().get_financial_statements(stock_code=stock_code, report_dates=report_dates)
# Get the latest current year report period for market data
latest_current_year_report = None
@ -322,7 +331,8 @@ async def get_china_financials(
has_daily_basic = bool(api_groups.get("daily_basic"))
has_daily = bool(api_groups.get("daily"))
if has_daily_basic or has_daily:
# 目前仅对中国市场启用 daily_basic/daily 数据拉取,其他市场由对应 provider 后续实现
if market == MarketEnum.cn and (has_daily_basic or has_daily):
step_market = StepRecord(name="拉取市值与股价", start_ts=datetime.now(timezone.utc).isoformat(), status="running")
steps.append(step_market)
@ -339,7 +349,7 @@ async def get_china_financials(
try:
if has_daily_basic:
db_rows = await get_dm().get_data('get_daily_basic_points', stock_code=ts_code, trade_dates=market_dates)
db_rows = await get_dm().get_data('get_daily_basic_points', stock_code=stock_code, trade_dates=market_dates)
if isinstance(db_rows, list):
for row in db_rows:
trade_date = row.get('trade_date') or row.get('trade_dt') or row.get('date')
@ -369,7 +379,7 @@ async def get_china_financials(
else:
series[key].append({"period": period, "value": value})
if has_daily:
d_rows = await get_dm().get_data('get_daily_points', stock_code=ts_code, trade_dates=market_dates)
d_rows = await get_dm().get_data('get_daily_points', stock_code=stock_code, trade_dates=market_dates)
if isinstance(d_rows, list):
for row in d_rows:
trade_date = row.get('trade_date') or row.get('trade_dt') or row.get('date')
@ -413,10 +423,23 @@ async def get_china_financials(
if not series:
raise HTTPException(status_code=502, detail={"message": "No data returned from any data source", "errors": errors})
# Truncate periods and sort (the data should already be mostly correct, but we ensure)
for key, arr in series.items():
# Deduplicate and sort desc by period, then cut to requested periods, and return asc
uniq = {item["period"]: item for item in arr}
# 统一 period 字段;若仅有 year 则映射为 YYYY1231然后去重与排序
for key, arr in list(series.items()):
normalized: List[Dict] = []
for item in arr:
period = item.get("period")
if not period:
year = item.get("year")
if year:
period = f"{str(year)}1231"
if not period:
# 跳过无法确定 period 的项
continue
value = item.get("value")
normalized.append({"period": str(period), "value": value})
# Deduplicate by period
uniq = {it["period"]: it for it in normalized}
arr_sorted_desc = sorted(uniq.values(), key=lambda x: x["period"], reverse=True)
arr_limited = arr_sorted_desc[:years]
arr_sorted = sorted(arr_limited, key=lambda x: x["period"])
@ -438,7 +461,7 @@ async def get_china_financials(
steps=steps,
)
return BatchFinancialDataResponse(ts_code=ts_code, name=company_name, series=series, meta=meta)
return BatchFinancialDataResponse(ts_code=stock_code, name=company_name, series=series, meta=meta)
@router.get("/china/{ts_code}/company-profile", response_model=CompanyProfileResponse)
@ -800,6 +823,66 @@ async def get_today_snapshot(ts_code: str):
raise HTTPException(status_code=500, detail=f"Failed to fetch snapshot: {e}")
@router.get("/{market}/{stock_code}/snapshot", response_model=TodaySnapshotResponse)
async def get_market_snapshot(market: MarketEnum, stock_code: str):
"""
市场无关的昨日快照接口
- CN: 复用中国市场的快照逻辑daily_basic/daily
- 其他市场: 兜底使用日行情获取最近交易日收盘价其余字段暂返回空值
"""
if market == MarketEnum.cn:
return await get_today_snapshot(stock_code)
try:
# 公司名称(可选)
company_name = None
try:
basic = await get_dm().get_stock_basic(stock_code=stock_code)
if basic:
company_name = basic.get("name")
except Exception:
company_name = None
base_dt = (datetime.now() - timedelta(days=1)).date()
base_str = base_dt.strftime("%Y%m%d")
# 为了稳妥拿到最近交易日,回看近 10 天
start_dt = base_dt - timedelta(days=10)
start_str = start_dt.strftime("%Y%m%d")
end_dt = base_dt + timedelta(days=1)
end_str = end_dt.strftime("%Y%m%d")
rows = await get_dm().get_daily_price(stock_code=stock_code, start_date=start_str, end_date=end_str)
trade_date = None
close = None
if isinstance(rows, list) and rows:
# 选择 <= base_str 的最后一条记录
try:
candidates = [r for r in rows if str(r.get("trade_date") or r.get("date") or "") <= base_str]
if candidates:
last = sorted(candidates, key=lambda r: str(r.get("trade_date") or r.get("date") or ""))[-1]
trade_date = str(last.get("trade_date") or last.get("date") or base_str)
close = last.get("close")
except Exception:
pass
if trade_date is None:
trade_date = base_str
return TodaySnapshotResponse(
ts_code=stock_code,
trade_date=trade_date,
name=company_name,
close=close,
pe=None,
pb=None,
dv_ratio=None,
total_mv=None,
)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Failed to fetch snapshot: {e}")
@router.get("/china/{ts_code}/analysis/{analysis_type}/stream")
async def stream_analysis(
ts_code: str,
@ -925,3 +1008,16 @@ async def stream_analysis(
"X-Accel-Buffering": "no",
}
return StreamingResponse(streamer(), media_type="text/plain; charset=utf-8", headers=headers)
@router.get("/{market}/{stock_code}/analysis/{analysis_type}/stream")
async def stream_analysis_market(
market: MarketEnum,
stock_code: str,
analysis_type: str,
company_name: str = Query(None, description="Company name for better context"),
):
"""
市场无关的分析流接口逻辑与中国市场一致仅路径不同
"""
return await stream_analysis(stock_code, analysis_type, company_name)

View File

@ -19,9 +19,9 @@
- [x] **T2.1 [Backend/DB]**: 根据设计文档使用SQLAlchemy ORM定义`Report`, `AnalysisModule`, `ProgressTracking`, `SystemConfig`四个核心数据模型。 **[完成 - 2025-10-21]**
- [x] **T2.2 [Backend/DB]**: 创建第一个Alembic迁移脚本在数据库中生成上述四张表。 **[完成 - 2025-10-21]**
- [x] **T2.3 [Backend]**: 实现`ConfigManager`服务,完成从`config.json`加载配置并与数据库配置合并的逻辑。 **[完成 - 2025-10-21]**
- **T2.4 [Backend/API]**: 创建Pydantic Schema用于配置接口的请求和响应 (`ConfigResponse`, `ConfigUpdateRequest`, `ConfigTestRequest`, `ConfigTestResponse`)。
- **T2.5 [Backend/API]**: 实现`/api/config`的`GET`和`PUT`端点,用于读取和更新系统配置。
- **T2.6 [Backend/API]**: 实现`/api/config/test`的`POST`端点,用于验证数据库连接等配置的有效性。
- [x] **T2.4 [Backend/API]**: 创建Pydantic Schema用于配置接口的请求和响应 (`ConfigResponse`, `ConfigUpdateRequest`, `ConfigTestRequest`, `ConfigTestResponse`)。
- [x] **T2.5 [Backend/API]**: 实现`/api/config`的`GET`和`PUT`端点,用于读取和更新系统配置。
- [x] **T2.6 [Backend/API]**: 实现`/api/config/test`的`POST`端点,用于验证数据库连接等配置的有效性。
## Phase 3: 前端基础与配置页面 (P1)
@ -39,8 +39,8 @@
此阶段是项目的核心,重点开发后端的报告生成流程和前端的实时进度展示。
- **T4.1 [Backend/Service]**: 实现`DataSourceManager`封装对Tushare和Yahoo Finance的数据获取逻辑。
- **T4.2 [Backend/Service]**: 实现`AIService`封装对Google Gemini API的调用逻辑包括Token使用统计。
- [x] **T4.1 [Backend/Service]**: 实现`DataSourceManager`封装对Tushare和Yahoo Finance的数据获取逻辑。
- [x] **T4.2 [Backend/Service]**: 实现`AIService`封装对Google Gemini API的调用逻辑包括Token使用统计。
- **T4.3 [Backend/Service]**: 实现`ProgressTracker`服务,提供`initialize`, `start_step`, `complete_step`, `get_progress`等方法,并与数据库交互。
- **T4.4 [Backend/Service]**: 定义`AnalysisModule`的基类/接口,并初步实现一到两个模块(如`FinancialDataModule`)作为示例。
- **T4.5 [Backend/Service]**: 实现核心的`ReportGenerator`服务,编排数据获取、各分析模块调用、进度更新的完整流程。

View File

@ -0,0 +1,67 @@
# 美国市场数据集成任务清单
本文档用于跟踪和管理为项目集成美国市场数据(使用 Finnhub 作为数据源)所需的各项开发任务。
## 任务列表
- [x] **后端:实现 FinnhubProvider 数据映射**
- **目标**:根据 `docs/financial_data_dictionary.md` 中的定义,在 `backend/app/data_providers/finnhub.py` 文件中,完成从 Finnhub API 原始数据到系统标准字段的完整映射。
- **关键点**
- [x] 处理直接映射的字段。
- [x] 实现所有需要通过计算得出的衍生指标。
- [x] 确保处理 `null` 或空值,避免计算错误。
- [x] 验证返回的数据结构符合 `DataManager` 的预期。
- [x] **后端:按市场分段的 API 路由**
- **目标**:在 `backend/app/routers/financial.py` 中,将现有的 `/api/v1/financials/china/{ts_code}` 改为按市场分段:`/api/v1/financials/{market}/{stock_code}`(示例:`/api/v1/financials/us/AAPL``/api/v1/financials/cn/600519.SH`)。
- **关键点**
- [x] 去除硬编码的 `china`,新增路径参数 `market`,并对取值做校验(`cn/us/hk/jp`)。
- [x] 使用单一处理函数,根据 `market` 分派到相应的数据提供方与代码格式规范。
- [x] **前端:更新 API 调用**
- **目标**:修改前端调用,基于用户选择的市场与股票代码,请求新的按市场分段路由。
- **关键点**
- [x] 替换 `useChinaFinancials`,新增通用 `useFinancials(market, stockCode, years)`
- [x] 将请求路径改为 `/api/financials/{market}/{stock_code}?years=...`(代理到后端对应的 `/api/v1/financials/{market}/{stock_code}`)。
- [ ] 确保展示与错误处理兼容美国、香港、日本等市场。
- [ ] **测试与验证**
- **目标**:对整个流程进行端到端测试,确保两个市场的功能都稳定可靠。
- **关键点**
- [ ] **中国市场回归测试**:使用多个中国 A 股代码测试,确保原有功能不受影响。
- [ ] **美国市场功能测试**:使用多个美国股票代码(如 `AAPL`, `MSFT`)测试,验证报告能否成功生成。
- [ ] **数据一致性验证**:抽样对比 Finnhub 返回的数据和前端展示的数据,确保映射和计算的准确性。
- [ ] **错误处理测试**:测试无效的股票代码,检查系统是否能给出清晰的错误提示。
- **前置条件**
- [ ] 在 `config/config.json` 或环境变量中配置 `FINNHUB_API_KEY`
- [ ] 后端已启动(默认 `http://127.0.0.1:8000/api`),前端已启动(默认 `http://127.0.0.1:3000`)。
- **接口用例(后端)**
- [ ] GET `/api/v1/financials/cn/600519.SH?years=10`
- 期望:`200`;返回 `ts_code`、`name`、`series`(含 `revenue`、`n_income` 等关键指标period/年序列齐全)。
- [ ] GET `/api/v1/financials/cn/000001.SZ?years=5`
- 期望:`200`;返回与上同,近 5 年序列。
- [ ] GET `/api/v1/financials/us/AAPL?years=10`
- 期望:`200``series` 至少包含:`revenue`、`n_income`、`total_assets`、`total_hldr_eqy_exc_min_int`、`__free_cash_flow`、`grossprofit_margin`、`netprofit_margin`、`roe`、`roa`。
- [ ] GET `/api/v1/financials/us/MSFT?years=10`
- 期望:`200`;字段与口径同 AAPL。
- [ ] GET `/api/v1/financials/us/INVALID?years=10`
- 期望:`4xx/5xx``detail.message` 含可读错误。
- **页面用例(前端)**
- [ ] 打开 `/report/600519.SH?market=cn`
- 期望:基本信息与“昨日快照”显示;“财务数据(来自 Tushare”表格展示 10 期内主要指标。
- [ ] 打开 `/report/000001.SZ?market=cn`
- 期望:与上同;代码规范化逻辑(无后缀时自动补 `.SZ/.SH`)正常。
- [ ] 打开 `/report/AAPL?market=us`
- 期望“股价图表”正常“财务数据”表格展示主要指标含自由现金流、毛利率、净利率、ROA、ROE
- [ ] 打开 `/report/MSFT?market=us`
- 期望:与上同。
- [ ] 打开 `/report/INVALID?market=us`
- 期望:顶部状态为“读取失败”并有错误提示文案。
- **验收标准**
- [ ] 中国市场功能无回归;美国市场关键指标齐全、值域合理(百分比类 ∈ [-1000%, 1000%],金额类为有限数)。
- [ ] 报错信息清晰可读;网络/密钥缺失时提示明确。
- [ ] 页内主要表格不出现 `NaN/Infinity`;空值以 `-` 展示。

View File

@ -18,3 +18,5 @@

View File

@ -1,7 +1,7 @@
'use client';
import { useParams, useSearchParams } from 'next/navigation';
import { useChinaFinancials, useFinancialConfig, useAnalysisConfig, generateFullAnalysis, useChinaSnapshot } from '@/hooks/useApi';
import { useChinaFinancials, useFinancials, useFinancialConfig, useAnalysisConfig, generateFullAnalysis, useSnapshot } from '@/hooks/useApi';
import { Spinner } from '@/components/ui/spinner';
import { Button } from '@/components/ui/button';
import { CheckCircle, XCircle, RotateCw } from 'lucide-react';
@ -20,11 +20,18 @@ export default function ReportPage() {
const params = useParams();
const searchParams = useSearchParams();
const symbol = params.symbol as string;
const market = (searchParams.get('market') || '').toLowerCase();
const marketParam = (searchParams.get('market') || '').toLowerCase();
const normalizedMarket = (() => {
if (marketParam === 'usa') return 'us';
if (marketParam === 'china') return 'cn';
if (marketParam === 'hkex') return 'hk';
if (marketParam === 'jpn') return 'jp';
return marketParam;
})();
const displayMarket = market === 'china' ? '中国' : market;
const displayMarket = marketParam === 'china' ? '中国' : marketParam;
const isChina = market === 'china' || market === 'cn';
const isChina = normalizedMarket === 'cn';
// 规范化中国市场 ts_code若为6位数字或无后缀自动推断交易所
const normalizedTsCode = (() => {
@ -40,8 +47,13 @@ export default function ReportPage() {
return symbol.toUpperCase();
})();
const { data: financials, error, isLoading } = useChinaFinancials(isChina ? normalizedTsCode : undefined, 10);
const { data: snapshot, error: snapshotError, isLoading: snapshotLoading } = useChinaSnapshot(isChina ? normalizedTsCode : undefined);
const chinaFin = useChinaFinancials(isChina ? normalizedTsCode : undefined, 10);
const otherFin = useFinancials(!isChina ? normalizedMarket : undefined, !isChina ? symbol : undefined, 10);
const financials = (chinaFin.data ?? otherFin.data) as any;
const error = chinaFin.error ?? otherFin.error;
const isLoading = chinaFin.isLoading || otherFin.isLoading;
const unifiedSymbol = isChina ? normalizedTsCode : symbol;
const { data: snapshot, error: snapshotError, isLoading: snapshotLoading } = useSnapshot(normalizedMarket, unifiedSymbol);
const { data: financialConfig } = useFinancialConfig();
const { data: analysisConfig } = useAnalysisConfig();
@ -139,7 +151,7 @@ export default function ReportPage() {
}
const runFullAnalysis = async () => {
if (!isChina || !financials || !analysisConfig?.analysis_modules || isAnalysisRunningRef.current) {
if (!financials || !analysisConfig?.analysis_modules || isAnalysisRunningRef.current) {
return;
}
@ -321,7 +333,7 @@ export default function ReportPage() {
}, [startTime]);
const retryAnalysis = async (analysisType: string) => {
if (!isChina || !financials || !analysisConfig?.analysis_modules) {
if (!financials || !analysisConfig?.analysis_modules) {
return;
}
analysisFetchedRefs.current[analysisType] = false;
@ -344,7 +356,7 @@ export default function ReportPage() {
try {
const startedMsLocal = Date.now();
const response = await fetch(
`/api/financials/china/${normalizedTsCode}/analysis/${analysisType}/stream?company_name=${encodeURIComponent(financials?.name || normalizedTsCode)}`
`/api/financials/${normalizedMarket}/${unifiedSymbol}/analysis/${analysisType}/stream?company_name=${encodeURIComponent(financials?.name || unifiedSymbol)}`
);
if (!response.ok) {
throw new Error(`HTTP error! status: ${response.status}`);
@ -420,7 +432,7 @@ export default function ReportPage() {
};
useEffect(() => {
if (!isChina || isLoading || error || !financials || !analysisConfig?.analysis_modules || analysisTypes.length === 0) {
if (isLoading || error || !financials || !analysisConfig?.analysis_modules || analysisTypes.length === 0) {
return;
}
if (isAnalysisRunningRef.current) {
@ -477,7 +489,7 @@ export default function ReportPage() {
abortControllerRef.current = new AbortController();
const startedMsLocal = Date.now();
const response = await fetch(
`/api/financials/china/${normalizedTsCode}/analysis/${analysisType}/stream?company_name=${encodeURIComponent(financials?.name || normalizedTsCode)}`,
`/api/financials/${normalizedMarket}/${unifiedSymbol}/analysis/${analysisType}/stream?company_name=${encodeURIComponent(financials?.name || unifiedSymbol)}`,
{ signal: abortControllerRef.current.signal }
);
if (!response.ok) {
@ -572,7 +584,7 @@ export default function ReportPage() {
}
};
runAnalysesSequentially();
}, [isChina, isLoading, error, financials, analysisConfig, analysisTypes, normalizedTsCode, manualRunKey]);
}, [isLoading, error, financials, analysisConfig, analysisTypes, normalizedTsCode, manualRunKey]);
const stopAll = () => {
stopRequestedRef.current = true;
@ -603,7 +615,7 @@ export default function ReportPage() {
<CardContent className="space-y-2 text-sm">
<div className="flex items-center gap-2">
<span className="text-muted-foreground min-w-20">:</span>
<span className="font-medium">{normalizedTsCode}</span>
<span className="font-medium">{unifiedSymbol}</span>
</div>
<div className="flex items-center gap-2">
<span className="text-muted-foreground min-w-20">:</span>
@ -627,8 +639,7 @@ export default function ReportPage() {
</CardContent>
</Card>
{isChina && (
<Card className="w-80 flex-shrink-0">
<Card className="w-80 flex-shrink-0">
<CardHeader>
<CardTitle className="text-lg"></CardTitle>
</CardHeader>
@ -714,9 +725,7 @@ export default function ReportPage() {
</div>
</CardContent>
</Card>
)}
{isChina && (
<Card className="w-40 flex-shrink-0">
<Card className="w-40 flex-shrink-0">
<CardContent className="flex flex-col gap-2">
<Button onClick={runFullAnalysis} disabled={isAnalysisRunningRef.current}>
{isAnalysisRunningRef.current ? '正在分析…' : '开始分析'}
@ -729,9 +738,7 @@ export default function ReportPage() {
</Button>
</CardContent>
</Card>
)}
{isChina && (
<Card className="w-80">
<Card className="w-80">
<CardHeader className="flex flex-col space-y-2 pb-2">
<div className="flex flex-row items-center justify-between w-full">
<CardTitle className="text-xl"></CardTitle>
@ -774,11 +781,9 @@ export default function ReportPage() {
)}
</CardContent>
</Card>
)}
</div>
{isChina && (
<Tabs defaultValue="chart" className="mt-4">
<Tabs defaultValue="chart" className="mt-4">
<TabsList className="flex-wrap">
<TabsTrigger value="chart"></TabsTrigger>
<TabsTrigger value="financial"></TabsTrigger>
@ -795,19 +800,19 @@ export default function ReportPage() {
<div className="flex items-center gap-3 text-sm mb-4">
<CheckCircle className="size-4 text-green-600" />
<div className="text-muted-foreground">
- {normalizedTsCode}
- {unifiedSymbol}
</div>
</div>
<TradingViewWidget
symbol={normalizedTsCode}
market={market}
symbol={unifiedSymbol}
market={marketParam}
height={500}
/>
</TabsContent>
<TabsContent value="financial" className="space-y-4">
<h2 className="text-lg font-medium"> Tushare</h2>
<h2 className="text-lg font-medium"></h2>
<div className="flex items-center gap-3 text-sm">
{isLoading ? (
<Spinner className="size-4" />
@ -817,11 +822,7 @@ export default function ReportPage() {
<CheckCircle className="size-4 text-green-600" />
)}
<div className="text-muted-foreground">
{isLoading
? '正在读取 Tushare 数据…'
: error
? '读取失败'
: '读取完成'}
{isLoading ? '正在读取数据…' : error ? '读取失败' : '读取完成'}
</div>
</div>
{error && <p className="text-red-500"></p>}
@ -1527,7 +1528,6 @@ export default function ReportPage() {
)}
</TabsContent>
</Tabs>
)}
</div>
);
}

View File

@ -63,7 +63,7 @@ export function useFinancialConfig() {
export function useChinaFinancials(ts_code?: string, years: number = 10) {
return useSWR<BatchFinancialDataResponse>(
ts_code ? `/api/financials/china/${encodeURIComponent(ts_code)}?years=${encodeURIComponent(String(years))}` : null,
ts_code ? `/api/financials/cn/${encodeURIComponent(ts_code)}?years=${encodeURIComponent(String(years))}` : null,
fetcher,
{
revalidateOnFocus: false, // 不在窗口聚焦时重新验证
@ -74,6 +74,28 @@ export function useChinaFinancials(ts_code?: string, years: number = 10) {
);
}
export function useFinancials(market?: string, stockCode?: string, years: number = 10) {
const normalizeMarket = (m?: string) => {
const t = (m || '').toLowerCase();
if (t === 'usa') return 'us';
if (t === 'china') return 'cn';
if (t === 'hkex') return 'hk';
if (t === 'jpn') return 'jp';
return t;
};
const mkt = normalizeMarket(market);
return useSWR<BatchFinancialDataResponse>(
mkt && stockCode ? `/api/financials/${encodeURIComponent(mkt)}/${encodeURIComponent(stockCode)}?years=${encodeURIComponent(String(years))}` : null,
fetcher,
{
revalidateOnFocus: false,
revalidateOnReconnect: false,
dedupingInterval: 300000,
errorRetryCount: 1,
}
);
}
export function useAnalysisConfig() {
return useSWR<AnalysisConfigResponse>('/api/financials/analysis-config', fetcher);
}
@ -124,3 +146,25 @@ export function useChinaSnapshot(ts_code?: string) {
}
);
}
export function useSnapshot(market?: string, stockCode?: string) {
const normalizeMarket = (m?: string) => {
const t = (m || '').toLowerCase();
if (t === 'usa') return 'us';
if (t === 'china') return 'cn';
if (t === 'hkex') return 'hk';
if (t === 'jpn') return 'jp';
return t;
};
const mkt = normalizeMarket(market);
return useSWR<TodaySnapshotResponse>(
mkt && stockCode ? `/api/financials/${encodeURIComponent(mkt)}/${encodeURIComponent(stockCode)}/snapshot` : null,
fetcher,
{
revalidateOnFocus: false,
revalidateOnReconnect: false,
dedupingInterval: 120000,
errorRetryCount: 1,
}
);
}

View File

@ -40,3 +40,5 @@ if (process.env.NODE_ENV !== 'production') globalForPrisma.prisma = prisma