mirror of
https://github.com/ZhuLinsen/daily_stock_analysis
synced 2026-09-20 10:53:33 +08:00
feat: add Futu OpenD as optional HK realtime and fundamental data source (#2269)
* feat: add Futu OpenD as an optional HK realtime and fundamental data source Add FutuFetcher and FutuFundamentalAdapter behind FUTU_OPEND_HOST/PORT, register the settings in Config and config_registry so the Web settings page can expose host, port and HK realtime priority, and route HK realtime quotes through a configurable futu/longbridge/akshare/yfinance order while keeping A-share priority untouched. Include offline tests for the adapter, config schema and HK routing/fallback, plus docs and CHANGELOG entries. * fix: wire Futu fundamentals into HK pipeline and restore quote supplementation - _fetch_offshore_fundamental_bundle() prefers the Futu fundamental adapter for HK when FUTU_OPEND_HOST is configured, and falls back to yfinance when Futu is absent or returns no usable content. - HK realtime priority loop now supplements missing quote fields (volume_ratio / turnover_rate / pe/pb / market cap) from later configured sources instead of returning after the first non-empty quote, matching the US path's _supplement_quote behavior. - capital_flow / boards blocks are filled from the Futu bundle for HK instead of being hard-coded not_supported; status and missing_fields aggregation updated accordingly. - Add regression tests for partial-quote supplementation and Futu fundamental bundle routing/fallback. * test: expect boards block ok when bundle provides belong_boards The Futu integration made the offshore boards block data-driven instead of hard-coded not_supported; update the existing US/HK fundamental context test to match (belong_boards from the bundle now surface as an ok boards block). * fix: preserve HK fallback_from metadata and normalize Futu quote timestamps - HK realtime priority loop now records the failed preferred source token and passes it as fallback_from when a later source takes over, so the pipeline and analysis context can mark the quote as degraded. - Futu snapshot update_time is a naive Beijing-time (UTC+8) string; attach the +08:00 offset before storing provider_timestamp so stale_seconds / is_stale / provider_timestamp freshness semantics are correct instead of being parsed as UTC. - Add regression tests for fallback_from propagation and timestamp normalization. * fix: normalize Futu belong_boards to name/type/code contract OpenD owner_plate returns plate_code / plate_name / plate_type, but DSA downstream consumers (notification, extract_board_detail_fields, market structure) only read name/type/code. Map the fields in FutuFundamentalAdapter._boards so HK Futu boards are actually consumed instead of silently dropped, and add regression tests including an end-to-end check through extract_board_detail_fields. * fix: merge yfinance bundle when Futu fundamental returns partial blocks Futu partial success (e.g. statements failed but static info worked) used to short-circuit the whole bundle, silently dropping the growth/earnings that the existing yfinance path could still provide. Now, when Futu returns content but is missing growth or earnings, fetch the yfinance bundle within the remaining budget and merge the missing blocks (growth/earnings/institution/capital_flow/belong_boards), keeping Futu-preferred values where both exist. Add regression test for the partial-success merge path. * fix: use field-level checks when deciding Futu-vs-yfinance growth/earnings The previous merge condition only checked dict truthiness, so a truthy growth/earnings shell (all-None core values or metadata-only keys such as report_date/period/currency) would skip the yfinance supplement and silently downgrade existing HK fundamentals. Add _earnings_block_has_values (a core numeric field or a populated dividend is required) and reuse the existing _has_meaningful_payload for growth; both the missing_core check and the merge loop now use these. Add regression test for the all-None-shell scenario. * fix: fill HK fundamental field gaps from yfinance instead of block-level checks Block-level meaningful checks still skipped the yfinance supplement when Futu hit only part of the growth/earnings fields (e.g. revenue_yoy but None net_profit_yoy, or earnings with only basic_eps), silently dropping fields the main branch used to provide. Replace the missing_core decision with a per-field gap list (growth: revenue_yoy/net_profit_yoy/gross_margin; earnings.financial_report: revenue/net_profit_parent/basic_eps/gross_profit) and make the merge field-level: keep Futu values, fill each missing field from yfinance. Add regression tests for partial-hit and all-None shells. * fix: normalize Futu dividends to the repo contract and treat dividend gaps as supplement triggers Futu OpenD dividend_list carries raw fields (statement/ex_date/record_date) which the notification/data_processing market-structure consumers do not read; the repo contract is ttm_cash_dividend_per_share, ttm_dividend_yield_pct and events[].cash_dividend_per_share / ex_dividend_date / event_date. Normalize events in FutuFundamentalAdapter._dividends_and_splits, compute TTM count/cash and yield from the latest quote, and teach _field_gaps/_merge_bundles to treat a dividend block that does not satisfy the contract as a gap so yfinance supplements it. Also dedupe FUTU_OPEND_HOST/PORT in full-guide_EN. * fix: read dividend yield price from UnifiedRealtimeQuote objects FutuFetcher.get_realtime_quote returns a UnifiedRealtimeQuote dataclass, not a dict, so the yield branch in _dividends_and_splits that guarded on isinstance(quote, dict) never ran on the live Futu path, silently dropping ttm_dividend_yield_pct while the contract check considered the dividend block complete. Read price via getattr(quote, 'price', None) and keep the dict fallback for other fetchers; add a regression test driving the real UnifiedRealtimeQuote shape. * fix: treat dividend blocks with TTM cash but no yield as supplement gaps The repo contract consumes ttm_cash_dividend_per_share and ttm_dividend_yield_pct together. When the Futu dividend path has events and TTM cash but the extra realtime price snapshot failed (quote None / no price), ttm_dividend_yield_pct cannot be computed and the block was previously treated as complete, so yfinance was never consulted and the notification rendered the yield as N/A. _dividend_contract_has_values() now requires the paired yield whenever TTM cash is present, so _field_gaps() triggers the yfinance supplement and _merge_bundles() replaces the incomplete dividend block. Add regression tests for the adapter-level gap shape (quote unavailable leaves no yield) and the manager-level supplement path (Futu cash without yield pulls yfinance and fills the yield). * fix: skip unconfigured Futu in HK realtime routing When FUTU_OPEND_HOST is not configured, the HK realtime priority loop used to still attempt the futu source, record it as the failed primary, and attach fallback_from='futu' to a successful quote from the next enabled source (longbridge/akshare/yfinance). Consumers then wrongly treated an enabled source's first success as degraded fallback data, contradicting the documented contract that Futu only participates when OpenD is configured. The HK loop now checks FutuFetcher.has_configured_endpoint() once and skips the futu token entirely when it is disabled, so no fallback_from is written. Existing configured-Futu routing tests explicitly patch the endpoint check; a new regression test asserts an unconfigured Futu is never called and the enriched quote carries fallback_from=None. * fix: release cached HK Futu fundamental fetcher in DataFetcherManager.close() The HK Futu fundamental path lazily creates and caches its own FutuFetcher (an OpenQuoteContext-backed OpenD connection) on _futu_fundamental_fetcher, but close() only released the TickFlow fetcher and the default fetchers snapshot. Explicit close / reload paths therefore left the OpenD connection hanging. close() now takes the cached _futu_fundamental_fetcher, clears the reference and calls its close() best-effort. A regression test injects an observable fetcher into _futu_fundamental_fetcher and asserts close() invokes it and clears the attribute. --------- Co-authored-by: BayMax local review <baymax-local@invalid>
This commit is contained in:
sunkai174634
@@ -13,6 +13,7 @@ STOCK_LIST=600519,300750,002594
|
||||
# futu-api 10.8 仅支持 IPv4；Docker 连接宿主机 OpenD 时请勿使用容器内的 127.0.0.1，详见 docs/full-guide.md。
|
||||
# FUTU_OPEND_HOST=127.0.0.1
|
||||
# FUTU_OPEND_PORT=11111
|
||||
# FUTU_HK_REALTIME_SOURCE_PRIORITY=futu,longbridge,akshare,yfinance
|
||||
# FUTU_SECURITY_FIRM=NONE # 可选；默认由 OpenD 自动识别，也可显式指定券商
|
||||
# FUTU_ACC_ID= # 可选；正整数，指定后只读取该真实账户
|
||||
|
||||
|
||||
@@ -351,6 +351,30 @@ const settingsHelpZhCN: SettingsHelpMap = {
|
||||
impact: ['影响现价、技术指标、盘中分析和部分报告字段。'],
|
||||
notes: ['单一数据源失败应降级到后续数据源，不应拖垮主流程。'],
|
||||
},
|
||||
'settings.data_source.FUTU_OPEND_HOST': {
|
||||
title: 'Futu OpenD 地址',
|
||||
summary: '配置 Futu OpenD 服务地址。留空时不启用 Futu 数据源。',
|
||||
usage: '填写 IPv4 地址或可解析到 IPv4 的主机名。',
|
||||
valueNotes: ['OpenD 必须允许 DSA 容器访问。'],
|
||||
impact: ['影响港股 Futu 实时行情、历史行情和基本面数据访问。'],
|
||||
notes: ['这是服务地址，不是 Futu 账号密码。'],
|
||||
},
|
||||
'settings.data_source.FUTU_OPEND_PORT': {
|
||||
title: 'Futu OpenD 端口',
|
||||
summary: '配置 Futu OpenD TCP 端口。',
|
||||
usage: '填写 1 到 65535 之间的端口，默认 11111。',
|
||||
valueNotes: ['端口必须与 OpenD 实际监听端口一致。'],
|
||||
impact: ['影响 DSA 与 Futu OpenD 的连接。'],
|
||||
notes: ['修改后通常需要重启 DSA 服务以重新建立连接。'],
|
||||
},
|
||||
'settings.data_source.FUTU_HK_REALTIME_SOURCE_PRIORITY': {
|
||||
title: 'Futu 港股实时数据源优先级',
|
||||
summary: '配置港股实时行情的 Futu/Longbridge/AkShare/Yfinance 尝试顺序。',
|
||||
usage: '使用英文逗号分隔，可选 futu、longbridge、akshare、yfinance。',
|
||||
valueNotes: ['前面的数据源优先尝试；失败后自动回退。'],
|
||||
impact: ['只影响港股实时行情，不改变 A 股实时数据源优先级。'],
|
||||
notes: ['未配置 Futu OpenD 时会自动跳过 futu。'],
|
||||
},
|
||||
'settings.data_source.search_api_keys': {
|
||||
title: '搜索服务 API Key',
|
||||
summary: '配置新闻与搜索增强所需的第三方搜索服务密钥。',
|
||||
@@ -1523,6 +1547,30 @@ const settingsHelpEnUS: SettingsHelpMap = {
|
||||
impact: ['Affects request count and per-request pressure for TickFlow batch prefetch.'],
|
||||
notes: ['This setting only affects TickFlow batch paths.'],
|
||||
},
|
||||
'settings.data_source.FUTU_OPEND_HOST': {
|
||||
title: 'Futu OpenD Host',
|
||||
summary: 'Configures the Futu OpenD service address. Leave empty to disable Futu.',
|
||||
usage: 'Use an IPv4 address or a hostname resolving to IPv4.',
|
||||
valueNotes: ['The OpenD service must be reachable from the DSA container.'],
|
||||
impact: ['Affects Futu HK realtime, historical, and fundamental data access.'],
|
||||
notes: ['This is a service address, not a Futu account credential.'],
|
||||
},
|
||||
'settings.data_source.FUTU_OPEND_PORT': {
|
||||
title: 'Futu OpenD Port',
|
||||
summary: 'Configures the Futu OpenD TCP port.',
|
||||
usage: 'Use a port from 1 to 65535; the default is 11111.',
|
||||
valueNotes: ['The port must match the OpenD listener.'],
|
||||
impact: ['Affects the DSA connection to Futu OpenD.'],
|
||||
notes: ['Restarting DSA is normally required after changing it.'],
|
||||
},
|
||||
'settings.data_source.FUTU_HK_REALTIME_SOURCE_PRIORITY': {
|
||||
title: 'Futu HK Realtime Source Priority',
|
||||
summary: 'Configures the Futu/Longbridge/AkShare/Yfinance order for HK realtime quotes.',
|
||||
usage: 'Use comma-separated futu, longbridge, akshare, or yfinance values.',
|
||||
valueNotes: ['Earlier providers are tried first; failures fall back automatically.'],
|
||||
impact: ['Affects HK realtime quotes only, not A-share realtime priority.'],
|
||||
notes: ['The futu entry is skipped when OpenD is not configured.'],
|
||||
},
|
||||
'settings.data_source.stock_index_remote': {
|
||||
title: 'Remote Stock Index',
|
||||
summary: 'Fetches the latest stock autocomplete index from GitHub main and caches it locally.',
|
||||
|
||||
@@ -626,6 +626,7 @@ class DataFetcherManager:
|
||||
"BaostockFetcher": {"cn"},
|
||||
"YfinanceFetcher": {"cn", "hk", "us", "jp", "kr", "tw"},
|
||||
"LongbridgeFetcher": {"hk", "us"},
|
||||
"FutuFetcher": {"hk"},
|
||||
"FinnhubFetcher": {"us"},
|
||||
"AlphaVantageFetcher": {"us"},
|
||||
}
|
||||
@@ -671,6 +672,7 @@ class DataFetcherManager:
|
||||
self._init_default_fetchers()
|
||||
self._fundamental_adapter = AkshareFundamentalAdapter()
|
||||
self._yfinance_fundamental_adapter = YfinanceFundamentalAdapter()
|
||||
self._futu_fundamental_fetcher = None
|
||||
self._tickflow_fetcher = None
|
||||
self._tickflow_api_key: Optional[str] = None
|
||||
self._tickflow_lock = RLock()
|
||||
@@ -1286,6 +1288,28 @@ class DataFetcherManager:
|
||||
except Exception as exc:
|
||||
logger.debug("[TickFlowFetcher] 关闭管理器资源失败: %s", exc)
|
||||
|
||||
# The HK Futu fundamental path lazily creates and caches its own
|
||||
# FutuFetcher (an OpenQuoteContext-backed connection) on
|
||||
# _futu_fundamental_fetcher; release it here so explicit close /
|
||||
# reload paths do not leak the OpenD connection.
|
||||
futu_fundamental_fetcher = getattr(self, "_futu_fundamental_fetcher", None)
|
||||
if futu_fundamental_fetcher is not None:
|
||||
self._futu_fundamental_fetcher = None
|
||||
close_futu = getattr(futu_fundamental_fetcher, "close", None)
|
||||
if callable(close_futu):
|
||||
try:
|
||||
close_futu()
|
||||
except Exception as exc:
|
||||
logger.debug("[FutuFetcher] 关闭管理器资源失败: %s", exc)
|
||||
|
||||
for fetcher in self._get_fetchers_snapshot():
|
||||
close = getattr(fetcher, "close", None)
|
||||
if callable(close):
|
||||
try:
|
||||
close()
|
||||
except Exception as exc:
|
||||
logger.debug("[%s] close failed: %s", fetcher.name, exc)
|
||||
|
||||
def __del__(self) -> None:
|
||||
try:
|
||||
self.close()
|
||||
@@ -1532,6 +1556,7 @@ class DataFetcherManager:
|
||||
from .baostock_fetcher import BaostockFetcher
|
||||
from .yfinance_fetcher import YfinanceFetcher
|
||||
from .longbridge_fetcher import LongbridgeFetcher
|
||||
from .futu_fetcher import FutuFetcher
|
||||
config = get_config()
|
||||
# 创建所有数据源实例（优先级在各 Fetcher 的 __init__ 中确定）
|
||||
efinance = EfinanceFetcher()
|
||||
@@ -1567,6 +1592,11 @@ class DataFetcherManager:
|
||||
else:
|
||||
logger.debug("[数据源初始化] 跳过未配置的 LongbridgeFetcher")
|
||||
|
||||
if FutuFetcher.has_configured_endpoint():
|
||||
optional_fetchers.append(FutuFetcher()) # 富途（港股，依赖 OpenD）
|
||||
else:
|
||||
logger.debug("[数据源初始化] 跳过未配置的 FutuFetcher")
|
||||
|
||||
finnhub_api_key = (getattr(config, "finnhub_api_key", None) or "").strip()
|
||||
if finnhub_api_key:
|
||||
from .finnhub_fetcher import FinnhubFetcher
|
||||
@@ -2061,6 +2091,7 @@ class DataFetcherManager:
|
||||
return "akshare_hk"
|
||||
mapping = {
|
||||
"LongbridgeFetcher": "longbridge",
|
||||
"FutuFetcher": "futu",
|
||||
"YfinanceFetcher": "yfinance",
|
||||
"AkshareFetcher": "akshare",
|
||||
"FinnhubFetcher": "finnhub",
|
||||
@@ -2173,12 +2204,76 @@ class DataFetcherManager:
|
||||
primary_kw: dict = {}
|
||||
secondary_kw: dict = {}
|
||||
else:
|
||||
primary_src = "LongbridgeFetcher" if prefer_lb else "AkshareFetcher"
|
||||
secondary_src = "AkshareFetcher" if prefer_lb else "LongbridgeFetcher"
|
||||
market_label = "港股"
|
||||
primary_kw = {"source": "hk"} if primary_src == "AkshareFetcher" else {}
|
||||
secondary_kw = {"source": "hk"} if secondary_src == "AkshareFetcher" else {}
|
||||
|
||||
hk_priority = [
|
||||
source.strip().lower()
|
||||
for source in getattr(
|
||||
config,
|
||||
"futu_hk_realtime_source_priority",
|
||||
"futu,longbridge,akshare,yfinance",
|
||||
).split(",")
|
||||
if source.strip()
|
||||
]
|
||||
source_map = {
|
||||
"futu": ("FutuFetcher", {}),
|
||||
"longbridge": ("LongbridgeFetcher", {}),
|
||||
"akshare": ("AkshareFetcher", {"source": "hk"}),
|
||||
"yfinance": ("YfinanceFetcher", {}),
|
||||
}
|
||||
primary_quote = None
|
||||
primary_token = None
|
||||
primary_src_index = -1
|
||||
fallback_from = None
|
||||
# Futu only participates when an OpenD endpoint is configured.
|
||||
# Skipping an unconfigured source here (instead of letting
|
||||
# _try_fetcher_quote fail on it) avoids recording a never-enabled
|
||||
# source as the failed primary, which would wrongly mark a
|
||||
# successful quote from the next enabled source as fallback.
|
||||
futu_enabled = False
|
||||
try:
|
||||
from data_provider.futu_fetcher import FutuFetcher
|
||||
futu_enabled = FutuFetcher.has_configured_endpoint()
|
||||
except Exception: # noqa: BLE001 - fail closed: treat futu as disabled
|
||||
futu_enabled = False
|
||||
for index, source in enumerate(hk_priority):
|
||||
mapped = source_map.get(source)
|
||||
if mapped is None:
|
||||
logger.warning("[实时行情] 忽略未知港股数据源: %s", source)
|
||||
continue
|
||||
fetcher_name, fetcher_kw = mapped
|
||||
if fetcher_name == "FutuFetcher" and not futu_enabled:
|
||||
logger.info(
|
||||
"[实时行情] 港股 %s 未配置 FUTU_OPEND_HOST，跳过 futu 源", stock_code
|
||||
)
|
||||
continue
|
||||
quote = self._try_fetcher_quote(stock_code, fetcher_name, **fetcher_kw)
|
||||
if quote is not None:
|
||||
primary_quote = quote
|
||||
primary_token = self._realtime_fetcher_token(fetcher_name, **fetcher_kw)
|
||||
primary_src_index = index
|
||||
logger.info("[实时行情] 港股 %s 成功获取 (来源: %s)", stock_code, fetcher_name)
|
||||
break
|
||||
# 该源失败：记住它的 token，供后续成功源作为 fallback_from 使用。
|
||||
if fallback_from is None:
|
||||
fallback_from = self._realtime_fetcher_token(fetcher_name, **fetcher_kw)
|
||||
if primary_quote is not None:
|
||||
# 用后续数据源补充缺失字段（volume_ratio / turnover_rate / 估值 / 市值），
|
||||
# 保持与美股路径一致的 _supplement_quote 补字段能力。
|
||||
for source in hk_priority[primary_src_index + 1:]:
|
||||
mapped = source_map.get(source)
|
||||
if mapped is None:
|
||||
continue
|
||||
if not self._quote_needs_supplement(primary_quote):
|
||||
break
|
||||
fetcher_name, fetcher_kw = mapped
|
||||
self._supplement_quote(stock_code, primary_quote, fetcher_name, **fetcher_kw)
|
||||
return self._enrich_realtime_quote(
|
||||
primary_quote,
|
||||
fallback_from=fallback_from,
|
||||
realtime_cache_ttl=getattr(config, "realtime_cache_ttl", None),
|
||||
)
|
||||
if log_final_failure:
|
||||
logger.info("[实时行情] 港股 %s 无可用数据源", stock_code)
|
||||
return None
|
||||
primary_token = self._realtime_fetcher_token(primary_src, **primary_kw)
|
||||
primary_quote = self._try_fetcher_quote(stock_code, primary_src, **primary_kw)
|
||||
fallback_from = primary_token if primary_quote is None else None
|
||||
@@ -2187,7 +2282,6 @@ class DataFetcherManager:
|
||||
primary_quote = self._supplement_quote(
|
||||
stock_code, primary_quote, secondary_src, **secondary_kw,
|
||||
)
|
||||
# 美股个股（非指数）尝试从 Finnhub/AlphaVantage 补充缺失字段
|
||||
if is_us and not is_us_index and primary_quote is not None:
|
||||
for extra_src in ["FinnhubFetcher", "AlphaVantageFetcher"]:
|
||||
primary_quote = self._supplement_quote(
|
||||
@@ -3143,6 +3237,66 @@ class DataFetcherManager:
|
||||
return False
|
||||
return True
|
||||
|
||||
@staticmethod
|
||||
def _dividend_contract_has_values(payload: Any) -> bool:
|
||||
"""Check whether the dividend block satisfies the repo contract.
|
||||
|
||||
Downstream consumers read ttm_cash_dividend_per_share /
|
||||
ttm_dividend_yield_pct and events[].cash_dividend_per_share /
|
||||
ex_dividend_date / event_date. Raw provider events that only carry
|
||||
provider-native keys (statement/ex_date/record_date) do not satisfy
|
||||
the contract, so they must be treated as missing.
|
||||
"""
|
||||
if not isinstance(payload, dict):
|
||||
return DataFetcherManager._has_meaningful_payload(payload)
|
||||
# The repo contract consumes ttm_cash_dividend_per_share and
|
||||
# ttm_dividend_yield_pct as a pair. A block with TTM cash but no
|
||||
# yield (e.g. the extra realtime price snapshot failed or returned
|
||||
# no price) is still missing a consumed field, so it must count as
|
||||
# a gap and be supplemented instead of being treated as complete.
|
||||
if DataFetcherManager._has_meaningful_payload(payload.get("ttm_cash_dividend_per_share")):
|
||||
return DataFetcherManager._has_meaningful_payload(payload.get("ttm_dividend_yield_pct"))
|
||||
for key in ("ttm_cash_dividend_per_share", "ttm_dividend_yield_pct"):
|
||||
if DataFetcherManager._has_meaningful_payload(payload.get(key)):
|
||||
return True
|
||||
events = payload.get("events")
|
||||
if isinstance(events, list):
|
||||
for event in events:
|
||||
if not isinstance(event, dict):
|
||||
continue
|
||||
if DataFetcherManager._has_meaningful_payload(
|
||||
event.get("cash_dividend_per_share")
|
||||
) or DataFetcherManager._has_meaningful_payload(
|
||||
event.get("ex_dividend_date") or event.get("event_date")
|
||||
):
|
||||
return True
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def _earnings_block_has_values(payload: Any) -> bool:
|
||||
"""Field-level check for the earnings block.
|
||||
|
||||
A truthy dict with only metadata (report_date/period/currency) is a
|
||||
shell, not usable earnings. Require a core numeric field (revenue /
|
||||
net_profit_parent / basic_eps / gross_profit) or a populated dividend
|
||||
section before treating the block as usable.
|
||||
"""
|
||||
if not isinstance(payload, dict):
|
||||
return DataFetcherManager._has_meaningful_payload(payload)
|
||||
report = payload.get("financial_report")
|
||||
if isinstance(report, dict):
|
||||
for key in ("revenue", "net_profit_parent", "basic_eps", "gross_profit"):
|
||||
if DataFetcherManager._has_meaningful_payload(report.get(key)):
|
||||
return True
|
||||
dividend = payload.get("dividend")
|
||||
if DataFetcherManager._has_meaningful_payload(dividend):
|
||||
return True
|
||||
# Fall back to the generic check for other earnings sub-blocks.
|
||||
for key in ("financial_reports", "indicators"):
|
||||
if key in payload and DataFetcherManager._has_meaningful_payload(payload.get(key)):
|
||||
return True
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def _infer_block_status(payload: Any, fallback_status: str) -> str:
|
||||
if DataFetcherManager._has_meaningful_payload(payload):
|
||||
@@ -3230,6 +3384,188 @@ class DataFetcherManager:
|
||||
**blocks,
|
||||
}
|
||||
|
||||
def _fetch_offshore_fundamental_bundle(
|
||||
self,
|
||||
stock_code: str,
|
||||
market: str,
|
||||
bundle_timeout: float,
|
||||
) -> Tuple[Dict[str, Any], Optional[str], int, str]:
|
||||
"""Fetch the fundamental bundle for offshore markets.
|
||||
|
||||
For HK with a configured Futu OpenD endpoint, try the Futu fundamental
|
||||
adapter first (company profile, statements, dividends/splits, capital
|
||||
flow, boards). When Futu succeeds only partially (e.g. statements
|
||||
failed but static info worked), fetch the yfinance bundle as well and
|
||||
merge the missing blocks so existing HK growth/earnings capability is
|
||||
never silently downgraded. Returns (payload, error, duration_ms,
|
||||
provider_name).
|
||||
"""
|
||||
from src.config import get_config
|
||||
|
||||
config = get_config()
|
||||
|
||||
def _use_yfinance() -> Tuple[Dict[str, Any], Optional[str], int, str]:
|
||||
payload, err, ms = self._run_with_retry(
|
||||
lambda: self._yfinance_fundamental_adapter.get_fundamental_bundle(stock_code),
|
||||
bundle_timeout,
|
||||
"fundamental_bundle_yfinance",
|
||||
)
|
||||
return payload or {}, err, ms, "fundamental_bundle_yfinance"
|
||||
|
||||
def _field_gaps(payload: Dict[str, Any]) -> List[str]:
|
||||
"""List core growth/earnings fields that are missing or value-less.
|
||||
|
||||
A field is a gap when it has no usable value, so the yfinance
|
||||
bundle should be consulted to fill it regardless of block-level
|
||||
truthiness.
|
||||
"""
|
||||
gaps: List[str] = []
|
||||
growth = payload.get("growth")
|
||||
if isinstance(growth, dict):
|
||||
for field in ("revenue_yoy", "net_profit_yoy", "gross_margin"):
|
||||
if not self._has_meaningful_payload(growth.get(field)):
|
||||
gaps.append(f"growth.{field}")
|
||||
earnings = payload.get("earnings")
|
||||
report = earnings.get("financial_report") if isinstance(earnings, dict) else None
|
||||
if isinstance(report, dict):
|
||||
for field in ("revenue", "net_profit_parent", "basic_eps", "gross_profit"):
|
||||
if not self._has_meaningful_payload(report.get(field)):
|
||||
gaps.append(f"earnings.financial_report.{field}")
|
||||
# Dividend: the repo contract consumes ttm_* fields and/or
|
||||
# events[].cash_dividend_per_share / ex_dividend_date. Raw OpenD
|
||||
# events (statement/ex_date/record_date) without normalization do
|
||||
# not satisfy it, so treat the block as a gap unless the contract
|
||||
# fields carry usable values.
|
||||
dividend = earnings.get("dividend") if isinstance(earnings, dict) else None
|
||||
if not DataFetcherManager._dividend_contract_has_values(dividend):
|
||||
gaps.append("earnings.dividend")
|
||||
return gaps
|
||||
|
||||
def _merge_bundles(
|
||||
futu_payload: Dict[str, Any],
|
||||
yfinance_payload: Dict[str, Any],
|
||||
futu_ms: int,
|
||||
yfinance_ms: int,
|
||||
) -> Tuple[Dict[str, Any], Optional[str], int, str]:
|
||||
"""Field-level merge: keep Futu values, fill gaps from yfinance."""
|
||||
merged: Dict[str, Any] = dict(futu_payload)
|
||||
|
||||
# growth: field-level fill.
|
||||
futu_growth = futu_payload.get("growth")
|
||||
yf_growth = yfinance_payload.get("growth")
|
||||
if isinstance(yf_growth, dict):
|
||||
growth = dict(futu_growth) if isinstance(futu_growth, dict) else {}
|
||||
for field, value in yf_growth.items():
|
||||
if not self._has_meaningful_payload(growth.get(field)):
|
||||
growth[field] = value
|
||||
if any(self._has_meaningful_payload(v) for v in growth.values()):
|
||||
merged["growth"] = growth
|
||||
|
||||
# earnings: financial_report field-level fill, dividend block-level.
|
||||
futu_earnings = futu_payload.get("earnings")
|
||||
yf_earnings = yfinance_payload.get("earnings")
|
||||
if isinstance(yf_earnings, dict):
|
||||
earnings = dict(futu_earnings) if isinstance(futu_earnings, dict) else {}
|
||||
futu_report = earnings.get("financial_report")
|
||||
yf_report = yf_earnings.get("financial_report")
|
||||
if isinstance(yf_report, dict):
|
||||
report = dict(futu_report) if isinstance(futu_report, dict) else {}
|
||||
for field, value in yf_report.items():
|
||||
if not self._has_meaningful_payload(report.get(field)):
|
||||
report[field] = value
|
||||
if any(self._has_meaningful_payload(v) for v in report.values()):
|
||||
earnings["financial_report"] = report
|
||||
if not DataFetcherManager._dividend_contract_has_values(
|
||||
earnings.get("dividend")
|
||||
) and DataFetcherManager._dividend_contract_has_values(
|
||||
yf_earnings.get("dividend")
|
||||
):
|
||||
earnings["dividend"] = yf_earnings.get("dividend")
|
||||
if any(
|
||||
DataFetcherManager._earnings_block_has_values(earnings.get(key))
|
||||
or self._has_meaningful_payload(earnings.get(key))
|
||||
for key in ("financial_report", "financial_reports", "dividend", "indicators")
|
||||
):
|
||||
merged["earnings"] = earnings
|
||||
|
||||
# Other blocks stay block-level (Futu wins, yfinance fills absent blocks).
|
||||
for key in ("institution", "capital_flow", "belong_boards"):
|
||||
if not self._has_meaningful_payload(merged.get(key)) and self._has_meaningful_payload(
|
||||
yfinance_payload.get(key)
|
||||
):
|
||||
merged[key] = yfinance_payload.get(key)
|
||||
|
||||
merged["source_chain"] = list(
|
||||
futu_payload.get("source_chain", [])
|
||||
) + list(yfinance_payload.get("source_chain", []))
|
||||
merged["errors"] = list(futu_payload.get("errors", [])) + list(
|
||||
yfinance_payload.get("errors", [])
|
||||
)
|
||||
has_content = any(
|
||||
merged.get(key)
|
||||
for key in ("growth", "earnings", "institution", "capital_flow", "belong_boards")
|
||||
)
|
||||
merged["status"] = "partial" if has_content else "not_supported"
|
||||
return merged, None, futu_ms + yfinance_ms, "fundamental_bundle_futu"
|
||||
|
||||
try:
|
||||
from data_provider.futu_fetcher import FutuFetcher
|
||||
from data_provider.futu_fundamental_adapter import FutuFundamentalAdapter
|
||||
except Exception as exc: # noqa: BLE001 - fail open to yfinance
|
||||
logger.warning("[futu-fundamental] import failed, using yfinance: %s", exc)
|
||||
return _use_yfinance()
|
||||
|
||||
if market != "hk" or not FutuFetcher.has_configured_endpoint():
|
||||
return _use_yfinance()
|
||||
|
||||
futu_fetcher = getattr(self, "_futu_fundamental_fetcher", None)
|
||||
if futu_fetcher is None:
|
||||
try:
|
||||
futu_fetcher = FutuFetcher()
|
||||
self._futu_fundamental_fetcher = futu_fetcher
|
||||
except Exception as exc: # noqa: BLE001 - fail open to yfinance
|
||||
logger.warning("[futu-fundamental] fetcher init failed, using yfinance: %s", exc)
|
||||
return _use_yfinance()
|
||||
|
||||
adapter = FutuFundamentalAdapter(futu_fetcher)
|
||||
futu_payload, futu_err, futu_ms = self._run_with_retry(
|
||||
lambda: adapter.get_fundamental_bundle(stock_code),
|
||||
bundle_timeout,
|
||||
"fundamental_bundle_futu",
|
||||
)
|
||||
if futu_err is None and isinstance(futu_payload, dict):
|
||||
has_content = any(
|
||||
futu_payload.get(key)
|
||||
for key in ("growth", "earnings", "institution", "capital_flow", "belong_boards")
|
||||
)
|
||||
if has_content:
|
||||
# Futu partial success: keep the blocks it returned but do not
|
||||
# silently drop any core growth/earnings field that yfinance
|
||||
# could still provide. Decide by field gaps (not block-level
|
||||
# truthiness) so partial Futu results are supplemented.
|
||||
gaps = _field_gaps(futu_payload)
|
||||
remaining_timeout = max(bundle_timeout - futu_ms / 1000.0, 0.0)
|
||||
if gaps and remaining_timeout > 0:
|
||||
yfinance_payload, yfinance_err, yfinance_ms = self._run_with_retry(
|
||||
lambda: self._yfinance_fundamental_adapter.get_fundamental_bundle(stock_code),
|
||||
remaining_timeout,
|
||||
"fundamental_bundle_yfinance",
|
||||
)
|
||||
if yfinance_err is None and isinstance(yfinance_payload, dict):
|
||||
return _merge_bundles(futu_payload, yfinance_payload, futu_ms, yfinance_ms)
|
||||
logger.warning(
|
||||
"[futu-fundamental] %s yfinance supplement failed (%s); keeping partial Futu bundle",
|
||||
stock_code,
|
||||
yfinance_err,
|
||||
)
|
||||
return futu_payload, None, futu_ms, "fundamental_bundle_futu"
|
||||
logger.info(
|
||||
"[futu-fundamental] %s bundle empty (status=%s), falling back to yfinance",
|
||||
stock_code,
|
||||
futu_payload.get("status"),
|
||||
)
|
||||
return _use_yfinance()
|
||||
|
||||
def _build_offshore_fundamental_context(
|
||||
self,
|
||||
stock_code: str,
|
||||
@@ -3322,22 +3658,26 @@ class DataFetcherManager:
|
||||
[valuation_err] if valuation_err else [],
|
||||
)
|
||||
|
||||
# Fundamental bundle via yfinance.
|
||||
# Fundamental bundle via Futu (HK only, when OpenD is configured), then
|
||||
# fall back to yfinance for the same payload shape.
|
||||
bundle_timeout = min(fetch_timeout, max(stage_timeout - (time.time() - start_ts), 0.0))
|
||||
if bundle_timeout <= 0:
|
||||
bundle_payload, bundle_err, bundle_ms = {}, "fundamental stage timeout", 0
|
||||
bundle_provider = "fundamental_bundle_yfinance"
|
||||
else:
|
||||
bundle_payload, bundle_err, bundle_ms = self._run_with_retry(
|
||||
lambda: self._yfinance_fundamental_adapter.get_fundamental_bundle(stock_code),
|
||||
bundle_timeout,
|
||||
"fundamental_bundle_yfinance",
|
||||
bundle_payload, bundle_err, bundle_ms, bundle_provider = (
|
||||
self._fetch_offshore_fundamental_bundle(
|
||||
stock_code,
|
||||
market,
|
||||
bundle_timeout,
|
||||
)
|
||||
)
|
||||
if not isinstance(bundle_payload, dict):
|
||||
bundle_payload = {}
|
||||
|
||||
bundle_chain = self._normalize_source_chain(
|
||||
bundle_payload.get("source_chain", []),
|
||||
"fundamental_bundle_yfinance",
|
||||
bundle_provider,
|
||||
str(bundle_payload.get("status", "not_supported")),
|
||||
bundle_ms,
|
||||
)
|
||||
@@ -3365,9 +3705,47 @@ class DataFetcherManager:
|
||||
list(adapter_errors),
|
||||
)
|
||||
|
||||
# capital_flow / dragon_tiger / boards: no offshore data feed today -> not_supported.
|
||||
for block in ("capital_flow", "dragon_tiger", "boards"):
|
||||
result_ctx[block] = self._build_fundamental_block(
|
||||
# capital_flow / dragon_tiger / boards: Futu fills capital_flow and
|
||||
# belong_boards for HK; everything else keeps not_supported (fail-open).
|
||||
futu_capital_flow = (
|
||||
bundle_payload.get("capital_flow")
|
||||
if isinstance(bundle_payload.get("capital_flow"), dict) and bundle_payload.get("capital_flow")
|
||||
else {}
|
||||
)
|
||||
if futu_capital_flow:
|
||||
result_ctx["capital_flow"] = self._build_fundamental_block(
|
||||
"ok" if futu_capital_flow.get("latest") or futu_capital_flow.get("rows") else "partial",
|
||||
futu_capital_flow,
|
||||
bundle_chain,
|
||||
[],
|
||||
)
|
||||
else:
|
||||
result_ctx["capital_flow"] = self._build_fundamental_block(
|
||||
"not_supported",
|
||||
{},
|
||||
[{"provider": "fundamental_pipeline", "result": "not_supported", "duration_ms": 0}],
|
||||
["not supported for this source"],
|
||||
)
|
||||
result_ctx["dragon_tiger"] = self._build_fundamental_block(
|
||||
"not_supported",
|
||||
{},
|
||||
[{"provider": "fundamental_pipeline", "result": "not_supported", "duration_ms": 0}],
|
||||
["not supported for offshore market"],
|
||||
)
|
||||
futu_boards = (
|
||||
bundle_payload.get("belong_boards")
|
||||
if isinstance(bundle_payload.get("belong_boards"), list) and bundle_payload.get("belong_boards")
|
||||
else []
|
||||
)
|
||||
if futu_boards:
|
||||
result_ctx["boards"] = self._build_fundamental_block(
|
||||
"ok",
|
||||
{"boards": futu_boards},
|
||||
bundle_chain,
|
||||
[],
|
||||
)
|
||||
else:
|
||||
result_ctx["boards"] = self._build_fundamental_block(
|
||||
"not_supported",
|
||||
{},
|
||||
[{"provider": "fundamental_pipeline", "result": "not_supported", "duration_ms": 0}],
|
||||
@@ -3443,27 +3821,34 @@ class DataFetcherManager:
|
||||
["not supported for offshore market"],
|
||||
)
|
||||
|
||||
result_ctx["belong_boards"] = belong_boards
|
||||
result_ctx["belong_boards"] = belong_boards or futu_boards
|
||||
|
||||
capital_flow_status = result_ctx["capital_flow"].get("status", "not_supported")
|
||||
boards_status = result_ctx["boards"].get("status", "not_supported")
|
||||
block_statuses = {
|
||||
"valuation": result_ctx["valuation"].get("status", "not_supported"),
|
||||
"growth": growth_status,
|
||||
"earnings": earnings_status,
|
||||
"institution": institution_status,
|
||||
"capital_flow": "not_supported",
|
||||
"capital_flow": capital_flow_status,
|
||||
"dragon_tiger": "not_supported",
|
||||
"boards": "not_supported",
|
||||
"boards": boards_status,
|
||||
}
|
||||
result_ctx["coverage"] = block_statuses
|
||||
for block in ("valuation", "growth", "earnings", "institution", "capital_flow", "dragon_tiger", "boards"):
|
||||
result_ctx["errors"].extend(result_ctx[block].get("errors", []))
|
||||
result_ctx["source_chain"].extend(result_ctx[block].get("source_chain", []))
|
||||
|
||||
active_statuses = {"valuation": valuation_status, "growth": growth_status, "earnings": earnings_status}
|
||||
active_statuses = {
|
||||
"valuation": valuation_status,
|
||||
"growth": growth_status,
|
||||
"earnings": earnings_status,
|
||||
"capital_flow": capital_flow_status,
|
||||
"boards": boards_status,
|
||||
}
|
||||
# tw institution (when present) counts toward the OVERALL status so a report that
|
||||
# only has 三大法人 data still surfaces fundamentals (consumers key off the top-level
|
||||
# status). missing_fields stays the original three blocks, so offshore markets
|
||||
# without institution data are byte-identical (institution is not_supported there).
|
||||
# status). Futu capital_flow / boards count the same way when they are available.
|
||||
status_values = list(active_statuses.values())
|
||||
if institution_status == "ok":
|
||||
status_values.append("ok")
|
||||
|
||||
349
data_provider/futu_fetcher.py
Normal file
349
data_provider/futu_fetcher.py
Normal file
@@ -0,0 +1,349 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""Futu OpenD market-data fetcher.
|
||||
|
||||
Read-only HK quote and daily-candlestick adapter for DSA. Trading APIs are
|
||||
intentionally not imported or exposed here.
|
||||
"""
|
||||
|
||||
import logging
|
||||
import os
|
||||
import threading
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import Any, Optional
|
||||
|
||||
import pandas as pd
|
||||
|
||||
from .base import BaseFetcher, STANDARD_COLUMNS
|
||||
from .realtime_types import RealtimeSource, UnifiedRealtimeQuote, safe_float
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Futu get_market_snapshot returns update_time as a naive "yyyy-MM-dd HH:mm:ss"
|
||||
# string; per the official docs HK and A-share quotes use Beijing time (UTC+8).
|
||||
# DSA's _parse_realtime_timestamp treats naive values as UTC, so we attach the
|
||||
# market-local offset here to keep provider_timestamp / stale_seconds correct.
|
||||
_HK_UPDATE_TIME_OFFSET = timezone(timedelta(hours=8))
|
||||
|
||||
|
||||
def _hk_provider_timestamp(value: Any) -> Optional[str]:
|
||||
"""Normalize Futu snapshot update_time to an offset-aware ISO string."""
|
||||
if value in (None, ""):
|
||||
return None
|
||||
text = str(value).strip()
|
||||
if not text:
|
||||
return None
|
||||
try:
|
||||
parsed = datetime.fromisoformat(text)
|
||||
except ValueError:
|
||||
return text
|
||||
if parsed.tzinfo is None:
|
||||
parsed = parsed.replace(tzinfo=_HK_UPDATE_TIME_OFFSET)
|
||||
return parsed.isoformat()
|
||||
|
||||
|
||||
def _hk_symbol(stock_code: str) -> Optional[str]:
|
||||
code = (stock_code or "").strip().upper()
|
||||
if code.endswith(".HK"):
|
||||
digits = code[:-3]
|
||||
elif code.startswith("HK"):
|
||||
digits = code[2:]
|
||||
elif code.isdigit() and 1 <= len(code) <= 5:
|
||||
digits = code
|
||||
else:
|
||||
return None
|
||||
if not digits.isdigit() or not 1 <= len(digits) <= 5:
|
||||
return None
|
||||
return f"HK.{digits.zfill(5)}"
|
||||
|
||||
|
||||
class FutuFetcher(BaseFetcher):
|
||||
"""Futu OpenD adapter for DSA's existing provider contract."""
|
||||
|
||||
name = "FutuFetcher"
|
||||
priority = int(os.getenv("FUTU_PRIORITY", "2"))
|
||||
allow_empty_daily_data = False
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._ctx = None
|
||||
self._ctx_lock = threading.Lock()
|
||||
self._available: Optional[bool] = None
|
||||
config = None
|
||||
try:
|
||||
from src.config import get_config
|
||||
config = get_config()
|
||||
except Exception:
|
||||
pass
|
||||
env_host = os.getenv("FUTU_OPEND_HOST")
|
||||
env_port = os.getenv("FUTU_OPEND_PORT")
|
||||
self._host = (getattr(config, "futu_opend_host", None) or env_host or "127.0.0.1").strip() or "127.0.0.1"
|
||||
try:
|
||||
self._port = int(getattr(config, "futu_opend_port", None) or env_port or 11111)
|
||||
except (TypeError, ValueError):
|
||||
self._port = 11111
|
||||
|
||||
@staticmethod
|
||||
def has_configured_endpoint() -> bool:
|
||||
if (os.getenv("FUTU_OPEND_HOST") or "").strip():
|
||||
return True
|
||||
try:
|
||||
from src.config import get_config
|
||||
return bool((getattr(get_config(), "futu_opend_host", None) or "").strip())
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
def _is_available(self) -> bool:
|
||||
if self._available is not None:
|
||||
return self._available
|
||||
try:
|
||||
import futu # noqa: F401
|
||||
self._available = self.has_configured_endpoint()
|
||||
except ImportError:
|
||||
self._available = False
|
||||
logger.warning("[Futu] futu-api 未安装")
|
||||
return self._available
|
||||
|
||||
def is_available_for_request(self, capability: str = "") -> bool:
|
||||
return self._is_available()
|
||||
|
||||
def _get_ctx(self):
|
||||
if not self._is_available():
|
||||
return None
|
||||
if self._ctx is not None:
|
||||
return self._ctx
|
||||
with self._ctx_lock:
|
||||
if self._ctx is None:
|
||||
try:
|
||||
from futu import OpenQuoteContext
|
||||
self._ctx = OpenQuoteContext(host=self._host, port=self._port)
|
||||
logger.info("[Futu] OpenQuoteContext 初始化成功: %s:%s", self._host, self._port)
|
||||
except Exception as exc:
|
||||
self._available = False
|
||||
logger.warning("[Futu] OpenQuoteContext 初始化失败: %s", exc)
|
||||
return self._ctx
|
||||
|
||||
@staticmethod
|
||||
def _close_ctx(ctx: Any) -> None:
|
||||
try:
|
||||
ctx.close()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def get_stock_name(self, stock_code: str) -> Optional[str]:
|
||||
symbol = _hk_symbol(stock_code)
|
||||
ctx = self._get_ctx()
|
||||
if not symbol or ctx is None:
|
||||
return None
|
||||
try:
|
||||
ret, data = ctx.get_market_snapshot([symbol])
|
||||
if ret == 0 and data is not None and not data.empty:
|
||||
return str(data.iloc[0].get("name") or "").strip() or None
|
||||
except Exception as exc:
|
||||
logger.debug("[Futu] stock name failed(%s): %s", symbol, exc)
|
||||
return None
|
||||
|
||||
def get_stock_basicinfo(self, stock_code: str):
|
||||
"""Return static security metadata for one HK symbol."""
|
||||
symbol = _hk_symbol(stock_code)
|
||||
ctx = self._get_ctx()
|
||||
if not symbol or ctx is None:
|
||||
return None
|
||||
try:
|
||||
import futu as ft
|
||||
ret, data = ctx.get_stock_basicinfo(ft.Market.HK, code_list=[symbol])
|
||||
return data if ret == ft.RET_OK else None
|
||||
except Exception as exc:
|
||||
logger.warning("[Futu] static info failed(%s): %s", symbol, exc)
|
||||
return None
|
||||
|
||||
def request_trading_days(self, start_date: str, end_date: str):
|
||||
"""Return HK trading days in the requested range."""
|
||||
ctx = self._get_ctx()
|
||||
if ctx is None:
|
||||
return []
|
||||
try:
|
||||
import futu as ft
|
||||
ret, data = ctx.request_trading_days(
|
||||
ft.TradeDateMarket.HK, start=start_date, end=end_date
|
||||
)
|
||||
return data if ret == ft.RET_OK else []
|
||||
except Exception as exc:
|
||||
logger.warning("[Futu] trading days failed: %s", exc)
|
||||
return []
|
||||
|
||||
def get_trading_days(self, start_date: str, end_date: str):
|
||||
"""Return normalized HK trading-day records."""
|
||||
rows = self.request_trading_days(start_date, end_date)
|
||||
return [
|
||||
{
|
||||
"date": row.get("time"),
|
||||
"trade_date_type": row.get("trade_date_type"),
|
||||
}
|
||||
for row in rows
|
||||
if isinstance(row, dict) and row.get("time")
|
||||
]
|
||||
|
||||
def get_company_profile(self, stock_code: str):
|
||||
symbol = _hk_symbol(stock_code)
|
||||
ctx = self._get_ctx()
|
||||
if not symbol or ctx is None:
|
||||
return None
|
||||
try:
|
||||
import futu as ft
|
||||
ret, data = ctx.get_company_profile(symbol)
|
||||
return data if ret == ft.RET_OK else None
|
||||
except Exception as exc:
|
||||
logger.warning("[Futu] company profile failed(%s): %s", symbol, exc)
|
||||
return None
|
||||
|
||||
def get_financials_statements(self, stock_code: str, statement_type: Optional[int] = None, num: int = 8):
|
||||
symbol = _hk_symbol(stock_code)
|
||||
ctx = self._get_ctx()
|
||||
if not symbol or ctx is None:
|
||||
return None
|
||||
try:
|
||||
import futu as ft
|
||||
ret, data = ctx.get_financials_statements(
|
||||
symbol, statement_type=statement_type, num=num
|
||||
)
|
||||
return data if ret == ft.RET_OK else None
|
||||
except Exception as exc:
|
||||
logger.warning("[Futu] financial statements failed(%s): %s", symbol, exc)
|
||||
return None
|
||||
|
||||
def get_corporate_actions_dividends(self, stock_code: str):
|
||||
symbol = _hk_symbol(stock_code)
|
||||
ctx = self._get_ctx()
|
||||
if not symbol or ctx is None:
|
||||
return None
|
||||
try:
|
||||
import futu as ft
|
||||
ret, data = ctx.get_corporate_actions_dividends(symbol)
|
||||
return data if ret == ft.RET_OK else None
|
||||
except Exception as exc:
|
||||
logger.warning("[Futu] dividends failed(%s): %s", symbol, exc)
|
||||
return None
|
||||
|
||||
def get_corporate_actions_stock_splits(self, stock_code: str, num: int = 50):
|
||||
symbol = _hk_symbol(stock_code)
|
||||
ctx = self._get_ctx()
|
||||
if not symbol or ctx is None:
|
||||
return None
|
||||
try:
|
||||
import futu as ft
|
||||
ret, data = ctx.get_corporate_actions_stock_splits(symbol, num=num)
|
||||
return data if ret == ft.RET_OK else None
|
||||
except Exception as exc:
|
||||
logger.warning("[Futu] stock splits failed(%s): %s", symbol, exc)
|
||||
return None
|
||||
|
||||
def get_capital_flow(self, stock_code: str, **kwargs):
|
||||
symbol = _hk_symbol(stock_code)
|
||||
ctx = self._get_ctx()
|
||||
if not symbol or ctx is None:
|
||||
return None
|
||||
try:
|
||||
import futu as ft
|
||||
ret, data = ctx.get_capital_flow(symbol, **kwargs)
|
||||
return data if ret == ft.RET_OK else None
|
||||
except Exception as exc:
|
||||
logger.warning("[Futu] capital flow failed(%s): %s", symbol, exc)
|
||||
return None
|
||||
|
||||
def get_owner_plate(self, stock_codes):
|
||||
symbols = [_hk_symbol(code) for code in stock_codes]
|
||||
symbols = [symbol for symbol in symbols if symbol]
|
||||
ctx = self._get_ctx()
|
||||
if not symbols or ctx is None:
|
||||
return None
|
||||
try:
|
||||
import futu as ft
|
||||
ret, data = ctx.get_owner_plate(symbols)
|
||||
return data if ret == ft.RET_OK else None
|
||||
except Exception as exc:
|
||||
logger.warning("[Futu] owner plate failed: %s", exc)
|
||||
return None
|
||||
|
||||
def close(self) -> None:
|
||||
if self._ctx is not None:
|
||||
self._close_ctx(self._ctx)
|
||||
self._ctx = None
|
||||
|
||||
def get_realtime_quote(self, stock_code: str) -> Optional[UnifiedRealtimeQuote]:
|
||||
symbol = _hk_symbol(stock_code)
|
||||
ctx = self._get_ctx()
|
||||
if not symbol or ctx is None:
|
||||
return None
|
||||
try:
|
||||
ret, rows = ctx.get_market_snapshot([symbol])
|
||||
if ret != 0 or rows is None or rows.empty:
|
||||
return None
|
||||
q = rows.iloc[0]
|
||||
price = safe_float(q.get("last_price"))
|
||||
if price is None or price <= 0:
|
||||
return None
|
||||
prev = safe_float(q.get("prev_close_price"))
|
||||
change_pct = safe_float(q.get("change_rate"))
|
||||
volume = int(q.get("volume") or 0)
|
||||
turnover = safe_float(q.get("turnover"))
|
||||
turnover_rate = safe_float(q.get("turnover_rate"))
|
||||
return UnifiedRealtimeQuote(
|
||||
code=stock_code,
|
||||
name=str(q.get("name") or ""),
|
||||
source=RealtimeSource.FUTU,
|
||||
price=price,
|
||||
change_pct=change_pct,
|
||||
change_amount=round(price - prev, 4) if prev else None,
|
||||
volume=volume or None,
|
||||
amount=turnover,
|
||||
turnover_rate=turnover_rate,
|
||||
volume_ratio=safe_float(q.get("volume_ratio")),
|
||||
amplitude=safe_float(q.get("amplitude")),
|
||||
open_price=safe_float(q.get("open_price")),
|
||||
high=safe_float(q.get("high_price")),
|
||||
low=safe_float(q.get("low_price")),
|
||||
pre_close=prev,
|
||||
pe_ratio=safe_float(q.get("pe_ttm_ratio")) or safe_float(q.get("pe_ratio")),
|
||||
pb_ratio=safe_float(q.get("pb_ratio")),
|
||||
total_mv=safe_float(q.get("total_market_val")),
|
||||
circ_mv=safe_float(q.get("circular_market_val")),
|
||||
provider_timestamp=_hk_provider_timestamp(q.get("update_time")),
|
||||
)
|
||||
except Exception as exc:
|
||||
logger.warning("[Futu] realtime quote failed(%s): %s", symbol, exc)
|
||||
return None
|
||||
|
||||
def _fetch_raw_data(self, stock_code: str, start_date: str, end_date: str) -> pd.DataFrame:
|
||||
symbol = _hk_symbol(stock_code)
|
||||
ctx = self._get_ctx()
|
||||
if not symbol or ctx is None:
|
||||
return pd.DataFrame()
|
||||
try:
|
||||
import futu as ft
|
||||
ret, data, _ = ctx.request_history_kline(
|
||||
symbol,
|
||||
start=start_date,
|
||||
end=end_date,
|
||||
ktype=ft.KLType.K_DAY,
|
||||
autype=ft.AuType.NONE,
|
||||
max_count=1000,
|
||||
)
|
||||
if ret != ft.RET_OK or data is None or data.empty:
|
||||
return pd.DataFrame()
|
||||
return data.rename(columns={"time_key": "date", "turnover": "amount"})
|
||||
except Exception as exc:
|
||||
logger.warning("[Futu] daily data failed(%s): %s", symbol, exc)
|
||||
return pd.DataFrame()
|
||||
|
||||
def _normalize_data(self, df: pd.DataFrame, stock_code: str) -> pd.DataFrame:
|
||||
if df.empty:
|
||||
return pd.DataFrame(columns=STANDARD_COLUMNS)
|
||||
out = df.copy()
|
||||
out["date"] = pd.to_datetime(out["date"]).dt.strftime("%Y-%m-%d")
|
||||
rename = {"open": "open", "high": "high", "low": "low", "close": "close", "volume": "volume"}
|
||||
out = out.rename(columns=rename)
|
||||
if "pct_chg" not in out.columns:
|
||||
out["pct_chg"] = out["close"].pct_change() * 100
|
||||
for col in STANDARD_COLUMNS:
|
||||
if col not in out.columns:
|
||||
out[col] = None
|
||||
return out[STANDARD_COLUMNS]
|
||||
276
data_provider/futu_fundamental_adapter.py
Normal file
276
data_provider/futu_fundamental_adapter.py
Normal file
@@ -0,0 +1,276 @@
|
||||
"""Futu fundamental data adapter for DSA offshore analysis."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
|
||||
import pandas as pd
|
||||
|
||||
|
||||
class FutuFundamentalAdapter:
|
||||
"""Normalize read-only Futu OpenD data to DSA's fundamental bundle."""
|
||||
|
||||
def __init__(self, fetcher: Any) -> None:
|
||||
self._fetcher = fetcher
|
||||
|
||||
@staticmethod
|
||||
def _ok_payload(result: Any) -> Tuple[Optional[Any], Optional[str]]:
|
||||
"""Unwrap the fetcher's payload-only contract."""
|
||||
if result is None:
|
||||
return None, "empty Futu response"
|
||||
return result, None
|
||||
|
||||
@staticmethod
|
||||
def _number(value: Any) -> Optional[float]:
|
||||
try:
|
||||
if value is None or pd.isna(value) or str(value).strip() in {"", "-", "N/A", "nan"}:
|
||||
return None
|
||||
return float(value)
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def _text(value: Any) -> Optional[str]:
|
||||
if value is None:
|
||||
return None
|
||||
text = str(value).strip()
|
||||
return text or None
|
||||
|
||||
@staticmethod
|
||||
def _code(stock_code: str) -> str:
|
||||
"""Keep DSA's internal HK code; FutuFetcher owns SDK conversion."""
|
||||
return stock_code
|
||||
|
||||
@staticmethod
|
||||
def _source(name: str, status: str = "ok") -> List[Dict[str, Any]]:
|
||||
return [{"provider": f"futu.{name}", "result": status, "duration_ms": 0}]
|
||||
|
||||
def _profile(self, code: str, result: Dict[str, Any]) -> None:
|
||||
static_info = self._fetcher.get_stock_basicinfo(code)
|
||||
if isinstance(static_info, pd.DataFrame) and not static_info.empty:
|
||||
row = static_info.iloc[0]
|
||||
result["institution"]["static_info"] = {
|
||||
key: row.get(key)
|
||||
for key in ("code", "name", "lot_size", "suspension", "listing_date", "exchange_type")
|
||||
if row.get(key) is not None
|
||||
}
|
||||
result["source_chain"].extend(self._source("static_info"))
|
||||
elif static_info is None:
|
||||
result["errors"].append("static_info:empty Futu response")
|
||||
|
||||
payload, error = self._ok_payload(self._fetcher.get_company_profile(code))
|
||||
if error:
|
||||
result["errors"].append(f"company_profile:{error}")
|
||||
return
|
||||
if isinstance(payload, pd.DataFrame) and not payload.empty:
|
||||
profile = {
|
||||
str(row["name"]): row["value"]
|
||||
for _, row in payload.iterrows()
|
||||
if "name" in row and "value" in row and self._text(row["name"])
|
||||
}
|
||||
if profile:
|
||||
result["institution"]["company_profile"] = profile
|
||||
result["source_chain"].extend(self._source("company_profile"))
|
||||
|
||||
def _financials(self, code: str, result: Dict[str, Any]) -> None:
|
||||
statements: Dict[str, Dict[str, Any]] = {}
|
||||
for statement_type, name in ((1, "income"), (2, "balance_sheet"), (3, "cash_flow"), (4, "indicators")):
|
||||
payload, error = self._ok_payload(
|
||||
self._fetcher.get_financials_statements(code, statement_type=statement_type, num=8)
|
||||
)
|
||||
if error:
|
||||
result["errors"].append(f"financials_{name}:{error}")
|
||||
continue
|
||||
if isinstance(payload, dict):
|
||||
statements[name] = payload
|
||||
|
||||
income = statements.get("income", {})
|
||||
reports = income.get("report_list") or []
|
||||
if not reports:
|
||||
return
|
||||
latest = reports[0] if isinstance(reports[0], dict) else {}
|
||||
items = {item.get("display_name"): item for item in latest.get("item_list", []) if isinstance(item, dict)}
|
||||
|
||||
def item(*names: str) -> Optional[Dict[str, Any]]:
|
||||
for name in names:
|
||||
if name in items:
|
||||
return items[name]
|
||||
return None
|
||||
|
||||
revenue = item("营业总收入", "营业额")
|
||||
net_profit = item("归属母公司净利润", "归属普通股股东净利润", "净利润")
|
||||
gross_profit = item("毛利")
|
||||
revenue_value = self._number((revenue or {}).get("data"))
|
||||
net_value = self._number((net_profit or {}).get("data"))
|
||||
gross_value = self._number((gross_profit or {}).get("data"))
|
||||
growth = {
|
||||
"revenue_yoy": self._number((revenue or {}).get("yoy")),
|
||||
"net_profit_yoy": self._number((net_profit or {}).get("yoy")),
|
||||
"gross_margin": (gross_value / revenue_value * 100.0) if gross_value is not None and revenue_value else None,
|
||||
}
|
||||
result["growth"].update({key: round(value, 6) if value is not None else None for key, value in growth.items()})
|
||||
|
||||
report = {
|
||||
"report_date": latest.get("date_time_str"),
|
||||
"period": latest.get("period_text"),
|
||||
"currency": latest.get("currency_code") or latest.get("currency_info"),
|
||||
"revenue": revenue_value,
|
||||
"net_profit_parent": net_value,
|
||||
"basic_eps": self._number((item("基本每股收益") or {}).get("data")),
|
||||
"gross_profit": gross_value,
|
||||
}
|
||||
for name, payload in statements.items():
|
||||
reports_for_type = payload.get("report_list") or []
|
||||
if reports_for_type:
|
||||
latest_items = reports_for_type[0].get("item_list", [])
|
||||
report[name] = {
|
||||
str(entry.get("display_name")): entry.get("data")
|
||||
for entry in latest_items
|
||||
if isinstance(entry, dict) and entry.get("data") is not None
|
||||
}
|
||||
result["earnings"]["financial_report"] = report
|
||||
result["earnings"]["financial_reports"] = {
|
||||
name: payload.get("report_list", []) for name, payload in statements.items()
|
||||
}
|
||||
result["source_chain"].extend(self._source("financials"))
|
||||
|
||||
def _dividends_and_splits(self, code: str, result: Dict[str, Any]) -> None:
|
||||
payload, error = self._ok_payload(self._fetcher.get_corporate_actions_dividends(code))
|
||||
if error:
|
||||
result["errors"].append(f"dividends:{error}")
|
||||
elif isinstance(payload, dict):
|
||||
raw_events = payload.get("dividend_list") or []
|
||||
events: List[Dict[str, Any]] = []
|
||||
ttm_events: List[Dict[str, Any]] = []
|
||||
ttm_cutoff = (datetime.now(timezone.utc) - timedelta(days=365)).date().isoformat()
|
||||
for raw in raw_events:
|
||||
if not isinstance(raw, dict):
|
||||
continue
|
||||
ex_date = self._text(raw.get("ex_date") or raw.get("ex_dividend_date"))
|
||||
per_share = self._number(raw.get("dividend_per_share"))
|
||||
if not ex_date and not per_share:
|
||||
continue
|
||||
# Normalize OpenD fields to the repo-wide dividend contract
|
||||
# consumed by notification / data_processing / market structure.
|
||||
event: Dict[str, Any] = {
|
||||
"event_date": ex_date or self._text(raw.get("record_date")),
|
||||
"ex_dividend_date": ex_date,
|
||||
"record_date": self._text(raw.get("record_date")),
|
||||
"announcement_date": self._text(raw.get("announcement_date")),
|
||||
"cash_dividend_per_share": per_share,
|
||||
"currency": self._text(raw.get("currency")),
|
||||
"statement": self._text(raw.get("statement")),
|
||||
"description": self._text(raw.get("description")),
|
||||
}
|
||||
if not any(v for v in (event["event_date"], event["cash_dividend_per_share"])):
|
||||
continue
|
||||
events.append(event)
|
||||
if event["event_date"] and event["event_date"] >= ttm_cutoff:
|
||||
ttm_events.append(event)
|
||||
events.sort(key=lambda item: item.get("event_date") or "", reverse=True)
|
||||
ttm_cash = (
|
||||
sum(float(item["cash_dividend_per_share"]) for item in ttm_events if item.get("cash_dividend_per_share") is not None)
|
||||
if ttm_events
|
||||
else None
|
||||
)
|
||||
dividend_payload: Dict[str, Any] = {
|
||||
"events": events[:5],
|
||||
"ttm_event_count": len(ttm_events),
|
||||
"ttm_cash_dividend_per_share": round(ttm_cash, 6) if ttm_cash is not None else None,
|
||||
"source": "futu",
|
||||
}
|
||||
if ttm_cash is not None:
|
||||
# Yield needs a price; try a lightweight snapshot if available.
|
||||
# FutuFetcher.get_realtime_quote returns a UnifiedRealtimeQuote
|
||||
# dataclass (not a dict), so read `price` via getattr to cover
|
||||
# both shapes.
|
||||
quote, quote_err = self._ok_payload(self._fetcher.get_realtime_quote(code))
|
||||
if not quote_err and quote is not None:
|
||||
latest_price = self._number(
|
||||
getattr(quote, "price", None)
|
||||
or (quote.get("price") if isinstance(quote, dict) else None)
|
||||
or (quote.get("last_price") if isinstance(quote, dict) else None)
|
||||
)
|
||||
if latest_price not in (None, 0):
|
||||
dividend_payload["ttm_dividend_yield_pct"] = round(
|
||||
float(ttm_cash) / float(latest_price) * 100.0, 4
|
||||
)
|
||||
result["earnings"]["dividend"] = dividend_payload
|
||||
result["source_chain"].extend(self._source("dividends"))
|
||||
|
||||
payload, error = self._ok_payload(self._fetcher.get_corporate_actions_stock_splits(code, num=50))
|
||||
if error:
|
||||
result["errors"].append(f"splits:{error}")
|
||||
elif isinstance(payload, dict):
|
||||
result["earnings"]["stock_splits"] = payload.get("split_list") or []
|
||||
result["source_chain"].extend(self._source("stock_splits"))
|
||||
|
||||
def _capital_flow(self, code: str, result: Dict[str, Any]) -> None:
|
||||
try:
|
||||
import futu
|
||||
period = futu.PeriodType.DAY
|
||||
except Exception:
|
||||
period = "DAY"
|
||||
payload, error = self._ok_payload(
|
||||
self._fetcher.get_capital_flow(code, period_type=period, start=None, end=None)
|
||||
)
|
||||
if error:
|
||||
result["errors"].append(f"capital_flow:{error}")
|
||||
elif isinstance(payload, pd.DataFrame) and not payload.empty:
|
||||
result["capital_flow"] = {
|
||||
"rows": payload.to_dict(orient="records"),
|
||||
"latest": payload.iloc[-1].to_dict(),
|
||||
}
|
||||
result["source_chain"].extend(self._source("capital_flow"))
|
||||
|
||||
def _boards(self, code: str, result: Dict[str, Any]) -> None:
|
||||
payload, error = self._ok_payload(self._fetcher.get_owner_plate([code]))
|
||||
if error:
|
||||
result["errors"].append(f"owner_plate:{error}")
|
||||
elif isinstance(payload, pd.DataFrame) and not payload.empty:
|
||||
# OpenD returns plate_code / plate_name / plate_type; DSA's downstream
|
||||
# consumers (notification, board-detail extraction, market structure)
|
||||
# only understand the name/type/code contract, so normalize here.
|
||||
boards = []
|
||||
for _, row in payload.iterrows():
|
||||
name = self._text(row.get("plate_name"))
|
||||
if not name:
|
||||
continue
|
||||
item: Dict[str, Any] = {"name": name}
|
||||
code_raw = self._text(row.get("plate_code"))
|
||||
if code_raw:
|
||||
item["code"] = code_raw
|
||||
type_raw = self._text(row.get("plate_type"))
|
||||
if type_raw:
|
||||
item["type"] = type_raw
|
||||
boards.append(item)
|
||||
result["belong_boards"] = boards
|
||||
result["source_chain"].extend(self._source("owner_plate"))
|
||||
|
||||
def get_fundamental_bundle(self, stock_code: str) -> Dict[str, Any]:
|
||||
code = self._code(stock_code)
|
||||
result: Dict[str, Any] = {
|
||||
"status": "not_supported",
|
||||
"growth": {},
|
||||
"earnings": {},
|
||||
"institution": {},
|
||||
"capital_flow": {},
|
||||
"belong_boards": [],
|
||||
"source_chain": [],
|
||||
"errors": [],
|
||||
}
|
||||
try:
|
||||
self._profile(code, result)
|
||||
self._financials(code, result)
|
||||
self._dividends_and_splits(code, result)
|
||||
self._capital_flow(code, result)
|
||||
self._boards(code, result)
|
||||
except Exception as exc:
|
||||
result["errors"].append(f"futu_adapter:{type(exc).__name__}:{exc}")
|
||||
has_content = any(
|
||||
result[key]
|
||||
for key in ("growth", "earnings", "institution", "capital_flow", "belong_boards")
|
||||
)
|
||||
result["status"] = "partial" if has_content else "not_supported"
|
||||
return result
|
||||
@@ -103,6 +103,7 @@ class RealtimeSource(Enum):
|
||||
SINA = "sina" # 新浪直连
|
||||
STOOQ = "stooq" # Stooq 美股兜底
|
||||
LONGBRIDGE = "longbridge" # 长桥（美股/港股兜底）
|
||||
FUTU = "futu" # 富途 OpenD（港股）
|
||||
FALLBACK = "fallback" # 降级兜底
|
||||
|
||||
|
||||
|
||||
@@ -14,6 +14,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
|
||||
- [新功能] Web Chat 意图识别层新增分词模块：`web_intent_tokenizer` 六步管道（多股票全名实体扫描 → 标点/空白切分 → 代码形提取 → 市场关键词 → 无歧义关键词 → 残存 gap 多策略 DFS 匹配）把用户消息切分为携带语义标签的 Token 序列；配套 `web_intent_types` 数据字典（Token 结构、Market 枚举、21 个语义 tag、clean/extend 双词池与正则机器）。核心原则"宁可不做，不可做错"：Step 1~5 只做精确匹配，Step 6 要求整段 TAG 全覆盖（交叉验证）才产出，未覆盖片段保持空 tag 交下游 LLM 兜底；代码形 token 辨认为 `stock_code`（附 code/name/market 三元组）/ `wrong_{market}_code` / `unknown_{market}_code` 三态，token 层代码拼写统一 canonical 归一（a=6 位裸数字、hk=HK+5 位、us=大写 ticker）。意图枚举与意图识别结果随后续 `web_intent_resolver` PR 引入。新增 183 个分词单元测试。
|
||||
<!-- 新条目格式：- [类型] 描述（类型取值：新功能/改进/修复/文档/测试/chore）-->
|
||||
<!-- 每条独立一行追加到本段末尾，无需分类标题，合并时冲突最小 -->
|
||||
- [新功能] 完善 Futu OpenD 港股数据源接入：系统设置支持 OpenD 地址、端口和港股实时数据源优先级，保留 Longbridge、AkShare、YFinance fallback。
|
||||
- [测试] 增加 Futu 配置 schema、港股实时路由和 fallback 契约覆盖。
|
||||
|
||||
- [新功能] 建立唯一、可生成、可校验、可降级的指数身份注册表：由 `scripts/stock_index_seeds/index_registry.csv` 的 31 项 manifest 确定性合并进 `apps/dsa-web/public/stocks.index.json`，运行时唯一真源为 JSON 中通过校验的 `active=true`/`assetType=index` 行，移除 `stock_list_parser` 的 5 项硬编码白名单；支持 `--index-only` 生成与字节稳定输出。
|
||||
- [新功能] 补齐显式 SH/SZ/CSI 指数 alias 收敛与 CSI 身份：`sh000300`/`000300.SH`/`sz399300`/`399300.SZ`/`000300.CSI` 均解析到 `sh000300`，`csi930955`/`930955.CSI` 解析到 `csi930955`；未登记 `.CSI` 输入返回 `unsupported`；裸数字恒为 stock 并仅通过 `matched_index` 暴露歧义。
|
||||
|
||||
@@ -11,7 +11,8 @@
|
||||
- A 股个股与选股：优先配置 `TUSHARE_TOKEN`，并保留 AkShare / Efinance / Tencent / TickFlow / Baostock / YFinance 兜底；普通个股日线按 priority 配置排序。
|
||||
- 已登记 A 股指数：固定按 Tencent → AkShare → TickFlow → YFinance 降级，不读取普通日 K 的 `*_PRIORITY` 配置。
|
||||
- A 股大盘复盘：配置 `TICKFLOW_API_KEY` 后，复盘聚合所需的指数和市场宽度会优先尝试 TickFlow，失败后回退现有免费源；这与单标的指数日线的 Tencent-first 固定链是不同入口。
|
||||
- 港股 / 美股：配置 `LONGBRIDGE_*` 后优先使用 Longbridge，YFinance、Finnhub、AlphaVantage 继续兜底。
|
||||
- 港股：配置 `FUTU_OPEND_HOST` 后，Futu 可作为港股实时与基本面主源；`FUTU_HK_REALTIME_SOURCE_PRIORITY` 控制港股实时行情顺序，Longbridge、AkShare、YFinance 保留为 fallback。
|
||||
- 美股：配置 `LONGBRIDGE_*` 后优先使用 Longbridge，YFinance、Finnhub、AlphaVantage 继续兜底。
|
||||
- 热点题材：选股的热点实现参考 AlphaSift，默认走 EastMoney provider，并使用本地 last-good cache 降低实时接口失败影响。
|
||||
|
||||
## 已接入数据源矩阵
|
||||
@@ -25,7 +26,8 @@
|
||||
| 选股快照 | Tushare、Sina、Efinance、AkShare EM、EastMoney Datacenter | 有 `TUSHARE_TOKEN` 时自动把 `tushare` 放入快照优先级；否则使用免费源链路 | 选股引擎维护 source health；状态接口透出 snapshot/daily health |
|
||||
| 选股日线补特征 | `DataFetcherManager` | 选股引擎优先复用现有日线与缓存链路 | 现有链路失败后才回到引擎自身的日线源 |
|
||||
| 选股热点题材 | EastMoney provider、参考 AlphaSift 的 hotspot 实现、last-good cache | 未指定 provider 时默认使用 EastMoney provider | 实时失败时回退热点缓存；无缓存时返回稳定空态和可读错误 |
|
||||
| 港股 / 美股 | Longbridge、YFinance、AkShare、Tushare、Finnhub、AlphaVantage、Stooq | 配置 Longbridge 凭证后参与港美股日线/实时兜底；YFinance 保持基础兜底 | Longbridge 冷却或失败时回退 YFinance / 其他可用源 |
|
||||
| 港股 | Futu、Longbridge、YFinance、AkShare、Tushare | 配置 `FUTU_OPEND_HOST` 后 Futu 作为实时与基本面主源，按 `FUTU_HK_REALTIME_SOURCE_PRIORITY` 顺序尝试 | Futu 失败时回退 Longbridge / AkShare / YFinance；Longbridge 冷却或失败时继续回退 YFinance / 其他可用源 |
|
||||
| 美股 | Longbridge、YFinance、AkShare、Tushare、Finnhub、AlphaVantage、Stooq | 配置 Longbridge 凭证后参与美股日线/实时兜底；YFinance 保持基础兜底 | Longbridge 冷却或失败时回退 YFinance / 其他可用源 |
|
||||
|
||||
## 总体链路图
|
||||
|
||||
@@ -43,7 +45,7 @@ flowchart TD
|
||||
C -->|缺失或过期| DM{市场}
|
||||
DM -->|A 股个股/未登记标的| CN[按 priority 动态排序: Efinance/AkShare/Tushare/TickFlow/Pytdx/Baostock/YFinance/Tencent]
|
||||
DM -->|已登记沪深指数| CNI[Tencent -> AkShare -> TickFlow -> YFinance]
|
||||
DM -->|港股| HK[Longbridge if configured -> AkShare/Tushare -> YFinance]
|
||||
DM -->|港股| HK[Futu if configured -> Longbridge/AkShare/YFinance fallback]
|
||||
DM -->|美股| US[Longbridge/YFinance -> Finnhub/AlphaVantage -> Stooq]
|
||||
|
||||
R --> RP[REALTIME_SOURCE_PRIORITY]
|
||||
|
||||
@@ -423,6 +423,7 @@ daily_stock_analysis/
|
||||
| `TICKFLOW_KLINE_ADJUST` | TickFlow 日 K 复权模式：`none`、`forward`、`backward`、`forward_additive`、`backward_additive`。 | `none` | 可选 |
|
||||
| `TICKFLOW_BATCH_DAILY_ENABLED` | 是否启用 TickFlow 批量日 K 预取；权限不足会短期缓存失败状态，并继续走常规回退。 | `true` | 可选 |
|
||||
| `TICKFLOW_BATCH_SIZE` | TickFlow 日 K 与实时行情批量请求的单批最大标的数。 | `100` | 可选 |
|
||||
| `FUTU_HK_REALTIME_SOURCE_PRIORITY` | 港股实时行情独立优先级，可选 `futu`、`longbridge`、`akshare`、`yfinance`，按逗号分隔；失败自动回退。 | `futu,longbridge,akshare,yfinance` | 可选 |
|
||||
| `LONGBRIDGE_OAUTH_CLIENT_ID` | Longbridge OAuth client_id；留空且无 Legacy Access Token 时会兼容使用 `LONGBRIDGE_APP_KEY` | - | 可选 |
|
||||
| `LONGBRIDGE_OAUTH_TOKEN_CACHE_B64` | OAuth token 缓存文件的 base64 内容，供 GitHub Actions / Docker 等 headless 环境使用 | - | 可选 |
|
||||
| `LONGBRIDGE_APP_KEY` | Longbridge Legacy App Key；无 `LONGBRIDGE_ACCESS_TOKEN` 时也可作为 OAuth client_id 兼容别名 | - | 可选 |
|
||||
|
||||
@@ -345,8 +345,6 @@ For the notification baseline, diagnostics, and deployment notes, see [Notificat
|
||||
|
||||
| Variable | Description | Default | Required |
|
||||
|--------|------|--------|:----:|
|
||||
| `FUTU_OPEND_HOST` | OpenD host. The pinned `futu-api==10.8.6808` accepts an IPv4 address or a hostname that resolves to IPv4. Cross-host connections should use only a trusted network or local port forwarding. | `127.0.0.1` | Optional |
|
||||
| `FUTU_OPEND_PORT` | OpenD port in the range `1-65535`. | `11111` | Optional |
|
||||
| `FUTU_SECURITY_FIRM` | Futu `SecurityFirm` enum name. `NONE` performs the SDK's official auto-detection once; set an explicit broker when required. | `NONE` | Optional |
|
||||
| `FUTU_ACC_ID` | Select one eligible REAL account ID. When empty, all explicitly `ACTIVE` `NORMAL` and `MASTER` securities accounts are merged. Treat account IDs as sensitive configuration and do not commit them. | empty | Optional |
|
||||
|
||||
@@ -363,7 +361,10 @@ For the notification baseline, diagnostics, and deployment notes, see [Notificat
|
||||
| `TENCENT_PRIORITY` | Tencent direct priority for the generic A-share daily K-line route; lower values are tried earlier and `5` is the default last fallback. Registered indices use a separate fixed chain and ignore this variable. Does not affect realtime quotes. | `5` | Optional |
|
||||
| `TICKFLOW_KLINE_ADJUST` | TickFlow daily K-line adjustment mode: `none`, `forward`, `backward`, `forward_additive`, or `backward_additive`. | `none` | Optional |
|
||||
| `TICKFLOW_BATCH_DAILY_ENABLED` | Enable TickFlow batch daily K-line prefetch when the current plan supports it; permission failures are negative-cached and fall back to per-stock providers. | `true` | Optional |
|
||||
| `TICKFLOW_BATCH_SIZE` | Maximum symbols per TickFlow batch request for daily K-lines and realtime quotes. | `100` | Optional |
|
||||
| `TICKFLOW_BATCH_SIZE` | Maximum symbols per TickFlow batch request. | `100` | Optional |
|
||||
| `FUTU_OPEND_HOST` | Futu OpenD address; use an IPv4 address or an IPv4-resolvable hostname. Leave empty to disable Futu market data. | empty | Optional |
|
||||
| `FUTU_OPEND_PORT` | Futu OpenD TCP port, from `1` to `65535`. | `11111` | Optional |
|
||||
| `FUTU_HK_REALTIME_SOURCE_PRIORITY` | HK realtime source order: `futu`, `longbridge`, `akshare`, or `yfinance`, comma-separated. Failed sources fall back automatically. | `futu,longbridge,akshare,yfinance` | Optional |
|
||||
| `ENABLE_REALTIME_QUOTE` | Enable real-time quotes (if disabled, uses historical closing prices for analysis) | `true` | Optional |
|
||||
| `ENABLE_REALTIME_TECHNICAL_INDICATORS` | Intraday real-time technicals: Calculate MA5/MA10/MA20 and bull trends using real-time prices when enabled (Issue #234); uses yesterday's close if disabled. | `true` | Optional |
|
||||
| `ENABLE_CHIP_DISTRIBUTION` | Enable chip distribution analysis (this API is unstable, recommended to disable for cloud deployment). GitHub Actions users must set `ENABLE_CHIP_DISTRIBUTION=true` in Repository Variables to enable; disabled by default in workflows. | `true` | Optional |
|
||||
|
||||
@@ -884,6 +884,9 @@ class Config:
|
||||
tickflow_priority: int = 2
|
||||
tickflow_batch_daily_enabled: bool = True
|
||||
tickflow_batch_size: int = 100
|
||||
futu_opend_host: Optional[str] = None
|
||||
futu_opend_port: int = 11111
|
||||
futu_hk_realtime_source_priority: str = "futu,longbridge,akshare,yfinance"
|
||||
finnhub_api_key: Optional[str] = None
|
||||
alphavantage_api_key: Optional[str] = None
|
||||
longbridge_app_key: Optional[str] = None
|
||||
@@ -1794,6 +1797,9 @@ class Config:
|
||||
tickflow_priority=parse_env_int(os.getenv('TICKFLOW_PRIORITY'), 2, field_name='TICKFLOW_PRIORITY', minimum=0),
|
||||
tickflow_batch_daily_enabled=parse_env_bool(os.getenv('TICKFLOW_BATCH_DAILY_ENABLED'), default=True),
|
||||
tickflow_batch_size=parse_env_int(os.getenv('TICKFLOW_BATCH_SIZE'), 100, field_name='TICKFLOW_BATCH_SIZE', minimum=1),
|
||||
futu_opend_host=os.getenv('FUTU_OPEND_HOST') or None,
|
||||
futu_opend_port=parse_env_int(os.getenv('FUTU_OPEND_PORT'), 11111, field_name='FUTU_OPEND_PORT', minimum=1, maximum=65535),
|
||||
futu_hk_realtime_source_priority=os.getenv('FUTU_HK_REALTIME_SOURCE_PRIORITY', 'futu,longbridge,akshare,yfinance'),
|
||||
finnhub_api_key=os.getenv('FINNHUB_API_KEY') or None,
|
||||
alphavantage_api_key=os.getenv('ALPHAVANTAGE_API_KEY') or None,
|
||||
longbridge_app_key=os.getenv('LONGBRIDGE_APP_KEY') or None,
|
||||
|
||||
@@ -848,6 +848,75 @@ _FIELD_DEFINITIONS: Dict[str, Dict[str, Any]] = {
|
||||
"validation": {"min": 1, "max": 500},
|
||||
"display_order": 19,
|
||||
},
|
||||
"FUTU_OPEND_HOST": {
|
||||
"title": "Futu OpenD Host",
|
||||
"description": "IPv4 address or IPv4-resolvable hostname of the Futu OpenD service. Leave empty to disable Futu.",
|
||||
"category": "data_source",
|
||||
"data_type": "string",
|
||||
"ui_control": "text",
|
||||
"is_sensitive": False,
|
||||
"is_required": False,
|
||||
"is_editable": True,
|
||||
"default_value": None,
|
||||
"options": [],
|
||||
"validation": {},
|
||||
"display_order": 20,
|
||||
"help_key": "settings.data_source.FUTU_OPEND_HOST",
|
||||
"examples": ["FUTU_OPEND_HOST=127.0.0.1"],
|
||||
"docs": [
|
||||
{
|
||||
"label": "数据源配置指南",
|
||||
"href": "https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/docs/full-guide.md#数据源配置",
|
||||
},
|
||||
],
|
||||
"warning_codes": [],
|
||||
},
|
||||
"FUTU_OPEND_PORT": {
|
||||
"title": "Futu OpenD Port",
|
||||
"description": "TCP port of the Futu OpenD service.",
|
||||
"category": "data_source",
|
||||
"data_type": "integer",
|
||||
"ui_control": "number",
|
||||
"is_sensitive": False,
|
||||
"is_required": False,
|
||||
"is_editable": True,
|
||||
"default_value": "11111",
|
||||
"options": [],
|
||||
"validation": {"min": 1, "max": 65535},
|
||||
"display_order": 21,
|
||||
"help_key": "settings.data_source.FUTU_OPEND_PORT",
|
||||
"examples": ["FUTU_OPEND_PORT=11111"],
|
||||
"docs": [
|
||||
{
|
||||
"label": "数据源配置指南",
|
||||
"href": "https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/docs/full-guide.md#数据源配置",
|
||||
},
|
||||
],
|
||||
"warning_codes": [],
|
||||
},
|
||||
"FUTU_HK_REALTIME_SOURCE_PRIORITY": {
|
||||
"title": "Futu 港股实时数据源优先级",
|
||||
"description": "港股实时行情优先级，可选 futu、longbridge、akshare、yfinance。未配置 OpenD 时自动跳过 futu。",
|
||||
"category": "data_source",
|
||||
"data_type": "string",
|
||||
"ui_control": "text",
|
||||
"is_sensitive": False,
|
||||
"is_required": False,
|
||||
"is_editable": True,
|
||||
"default_value": "futu,longbridge,akshare,yfinance",
|
||||
"options": [],
|
||||
"validation": {},
|
||||
"display_order": 22,
|
||||
"help_key": "settings.data_source.FUTU_HK_REALTIME_SOURCE_PRIORITY",
|
||||
"examples": ["FUTU_HK_REALTIME_SOURCE_PRIORITY=futu,longbridge,akshare,yfinance"],
|
||||
"docs": [
|
||||
{
|
||||
"label": "数据源配置指南",
|
||||
"href": "https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/docs/full-guide.md#数据源配置",
|
||||
},
|
||||
],
|
||||
"warning_codes": ["provider_priority_order"],
|
||||
},
|
||||
"STOCK_INDEX_REMOTE_UPDATE_ENABLED": {
|
||||
"title": "Remote Stock Index Updates",
|
||||
"description": "Automatically refresh the local stock autocomplete index from the built-in GitHub main source.",
|
||||
|
||||
@@ -879,5 +879,28 @@ class TestDingTalkWebhookFieldsRegistered(unittest.TestCase):
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
|
||||
class TestFutuFieldsRegistered(unittest.TestCase):
|
||||
def test_futu_fields_are_explicitly_registered(self):
|
||||
for key in (
|
||||
"FUTU_OPEND_HOST",
|
||||
"FUTU_OPEND_PORT",
|
||||
"FUTU_HK_REALTIME_SOURCE_PRIORITY",
|
||||
):
|
||||
field = get_field_definition(key)
|
||||
self.assertEqual(field["category"], "data_source")
|
||||
self.assertNotEqual(field["display_order"], 9000)
|
||||
|
||||
def test_futu_port_has_bounds(self):
|
||||
field = get_field_definition("FUTU_OPEND_PORT")
|
||||
self.assertEqual(field["data_type"], "integer")
|
||||
self.assertEqual(field["validation"], {"min": 1, "max": 65535})
|
||||
|
||||
def test_schema_response_includes_futu_fields(self):
|
||||
keys = {
|
||||
field["key"]
|
||||
for category in build_schema_response()["categories"]
|
||||
for field in category["fields"]
|
||||
}
|
||||
self.assertTrue({"FUTU_OPEND_HOST", "FUTU_OPEND_PORT", "FUTU_HK_REALTIME_SOURCE_PRIORITY"} <= keys)
|
||||
|
||||
@@ -138,13 +138,15 @@ class TestFundamentalContext(unittest.TestCase):
|
||||
):
|
||||
ctx = manager.get_fundamental_context("AAPL")
|
||||
self.assertEqual(ctx["market"], "us")
|
||||
# Offshore status only considers valuation/growth/earnings (capital_flow
|
||||
# etc. are intentionally not_supported); "ok" when all three populate.
|
||||
# Offshore status considers valuation/growth/earnings plus any populated
|
||||
# capital_flow / boards blocks; "ok" when the populated blocks are ok.
|
||||
self.assertEqual(ctx["status"], "ok")
|
||||
self.assertEqual(ctx["coverage"].get("growth"), "ok")
|
||||
self.assertEqual(ctx["coverage"].get("earnings"), "ok")
|
||||
self.assertEqual(ctx["coverage"].get("capital_flow"), "not_supported")
|
||||
self.assertEqual(ctx["coverage"].get("boards"), "not_supported")
|
||||
# belong_boards from the bundle surface the boards block (was hard-coded
|
||||
# not_supported before the Futu integration made it data-driven).
|
||||
self.assertEqual(ctx["coverage"].get("boards"), "ok")
|
||||
growth_data = ctx["growth"].get("data") or {}
|
||||
self.assertEqual(growth_data.get("revenue_yoy"), 16.5)
|
||||
self.assertEqual(growth_data.get("roe"), 141.4)
|
||||
|
||||
27
tests/test_futu_fetcher.py
Normal file
27
tests/test_futu_fetcher.py
Normal file
@@ -0,0 +1,27 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""Tests for FutuFetcher provider timestamp normalization."""
|
||||
|
||||
import unittest
|
||||
|
||||
from data_provider.futu_fetcher import _hk_provider_timestamp
|
||||
|
||||
|
||||
class TestFutuProviderTimestamp(unittest.TestCase):
|
||||
"""Futu snapshot update_time is a naive Beijing-time string."""
|
||||
|
||||
def test_naive_string_gets_beijing_offset(self):
|
||||
value = _hk_provider_timestamp("2026-08-24 10:30:00")
|
||||
self.assertEqual(value, "2026-08-24T10:30:00+08:00")
|
||||
|
||||
def test_empty_and_none_return_none(self):
|
||||
self.assertIsNone(_hk_provider_timestamp(None))
|
||||
self.assertIsNone(_hk_provider_timestamp(""))
|
||||
self.assertIsNone(_hk_provider_timestamp(" "))
|
||||
|
||||
def test_offset_string_is_preserved(self):
|
||||
value = _hk_provider_timestamp("2026-08-24T10:30:00+08:00")
|
||||
self.assertEqual(value, "2026-08-24T10:30:00+08:00")
|
||||
|
||||
def test_unparsable_text_returns_original(self):
|
||||
value = _hk_provider_timestamp("not-a-date")
|
||||
self.assertEqual(value, "not-a-date")
|
||||
673
tests/test_futu_fundamental_adapter.py
Normal file
673
tests/test_futu_fundamental_adapter.py
Normal file
@@ -0,0 +1,673 @@
|
||||
import unittest
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
import pandas as pd
|
||||
|
||||
from data_provider.futu_fundamental_adapter import FutuFundamentalAdapter
|
||||
|
||||
|
||||
class TestFutuFundamentalAdapter(unittest.TestCase):
|
||||
def _fetcher(self):
|
||||
fetcher = Mock()
|
||||
fetcher.get_stock_basicinfo.return_value = pd.DataFrame(
|
||||
[["HK.01810", "测试公司-W", 200, False, "2018-07-09", "HK_MAINBOARD"]],
|
||||
columns=["code", "name", "lot_size", "suspension", "listing_date", "exchange_type"],
|
||||
)
|
||||
fetcher.get_company_profile.return_value = pd.DataFrame(
|
||||
[["公司名称", "测试公司", 0]],
|
||||
columns=["name", "value", "field_type"],
|
||||
)
|
||||
fetcher.get_financials_statements.return_value = {
|
||||
"report_list": [
|
||||
{
|
||||
"date_time_str": "2026-06-30",
|
||||
"period_text": "2026/Q2",
|
||||
"currency_code": "CNY",
|
||||
"item_list": [
|
||||
{"display_name": "营业总收入", "data": 1000.0, "yoy": 10.0},
|
||||
{"display_name": "归属母公司净利润", "data": 200.0, "yoy": 20.0},
|
||||
{"display_name": "毛利", "data": 400.0, "yoy": 15.0},
|
||||
{"display_name": "基本每股收益", "data": 0.2},
|
||||
],
|
||||
}
|
||||
]
|
||||
}
|
||||
fetcher.get_corporate_actions_dividends.return_value = {"dividend_list": []}
|
||||
fetcher.get_corporate_actions_stock_splits.return_value = {"split_list": []}
|
||||
fetcher.get_capital_flow.return_value = pd.DataFrame(
|
||||
[{"capital_flow_item_time": "2026-08-21", "main_in_flow": 10.0}]
|
||||
)
|
||||
fetcher.get_owner_plate.return_value = pd.DataFrame(
|
||||
[{"plate_code": "HK.TEST", "plate_name": "测试行业", "plate_type": "INDUSTRY"}]
|
||||
)
|
||||
return fetcher
|
||||
|
||||
def test_normalizes_all_supported_blocks(self):
|
||||
fetcher = self._fetcher()
|
||||
bundle = FutuFundamentalAdapter(fetcher).get_fundamental_bundle("HK01810")
|
||||
|
||||
self.assertEqual(bundle["status"], "partial")
|
||||
self.assertEqual(bundle["growth"]["revenue_yoy"], 10.0)
|
||||
self.assertEqual(bundle["growth"]["gross_margin"], 40.0)
|
||||
self.assertEqual(bundle["earnings"]["financial_report"]["net_profit_parent"], 200.0)
|
||||
self.assertEqual(bundle["institution"]["company_profile"]["公司名称"], "测试公司")
|
||||
self.assertEqual(bundle["capital_flow"]["latest"]["main_in_flow"], 10.0)
|
||||
self.assertEqual(bundle["belong_boards"][0]["name"], "测试行业")
|
||||
self.assertEqual(bundle["belong_boards"][0]["code"], "HK.TEST")
|
||||
self.assertEqual(bundle["belong_boards"][0]["type"], "INDUSTRY")
|
||||
self.assertEqual(bundle["institution"]["static_info"]["lot_size"], 200)
|
||||
self.assertFalse(bundle["institution"]["static_info"]["suspension"])
|
||||
self.assertEqual(bundle["institution"]["company_profile"]["公司名称"], "测试公司")
|
||||
self.assertEqual(fetcher.get_financials_statements.call_count, 4)
|
||||
called_types = {
|
||||
call.kwargs["statement_type"]
|
||||
for call in fetcher.get_financials_statements.call_args_list
|
||||
}
|
||||
self.assertEqual(called_types, {1, 2, 3, 4})
|
||||
|
||||
def test_empty_corporate_actions_are_supported_empty_data(self):
|
||||
fetcher = self._fetcher()
|
||||
fetcher.request_trading_days.return_value = [
|
||||
{"time": "2026-08-24", "trade_date_type": "WHOLE"}
|
||||
]
|
||||
bundle = FutuFundamentalAdapter(fetcher).get_fundamental_bundle("HK01810")
|
||||
|
||||
self.assertEqual(bundle["earnings"]["dividend"]["events"], [])
|
||||
self.assertEqual(bundle["earnings"]["stock_splits"], [])
|
||||
self.assertNotIn("dividends:", " ".join(bundle["errors"]))
|
||||
self.assertNotIn("splits:", " ".join(bundle["errors"]))
|
||||
|
||||
def test_normalizes_trading_days(self):
|
||||
from data_provider.futu_fetcher import FutuFetcher
|
||||
|
||||
fetcher = FutuFetcher()
|
||||
fetcher.request_trading_days = Mock(
|
||||
return_value=[
|
||||
{"time": "2026-08-24", "trade_date_type": "WHOLE"},
|
||||
{"time": "", "trade_date_type": "WHOLE"},
|
||||
{"bad": True},
|
||||
]
|
||||
)
|
||||
self.assertEqual(
|
||||
fetcher.get_trading_days("2026-08-24", "2026-08-24"),
|
||||
[{"date": "2026-08-24", "trade_date_type": "WHOLE"}],
|
||||
)
|
||||
|
||||
def test_one_endpoint_failure_does_not_discard_other_blocks(self):
|
||||
fetcher = self._fetcher()
|
||||
fetcher.get_financials_statements.return_value = None
|
||||
bundle = FutuFundamentalAdapter(fetcher).get_fundamental_bundle("HK01810")
|
||||
|
||||
self.assertEqual(bundle["institution"]["company_profile"]["公司名称"], "测试公司")
|
||||
self.assertTrue(any(error.startswith("financials_") for error in bundle["errors"]))
|
||||
self.assertEqual(bundle["status"], "partial")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
|
||||
class TestFutuFundamentalIntegration(unittest.TestCase):
|
||||
"""Ensure get_fundamental_context() hits the Futu bundle for HK when configured."""
|
||||
|
||||
def _make_manager(self):
|
||||
import sys
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
if "litellm" not in sys.modules:
|
||||
sys.modules["litellm"] = MagicMock()
|
||||
if "json_repair" not in sys.modules:
|
||||
sys.modules["json_repair"] = MagicMock()
|
||||
|
||||
from data_provider.base import DataFetcherManager
|
||||
|
||||
manager = DataFetcherManager.__new__(DataFetcherManager)
|
||||
manager._futu_fundamental_fetcher = None
|
||||
manager._yfinance_fundamental_adapter = Mock()
|
||||
manager._yfinance_fundamental_adapter.get_fundamental_bundle.return_value = {
|
||||
"status": "not_supported", "growth": {}, "earnings": {},
|
||||
"belong_boards": [], "source_chain": [], "errors": [],
|
||||
}
|
||||
manager._fundamental_adapter = Mock()
|
||||
manager._fundamental_cache = {}
|
||||
manager._fundamental_cache_lock = __import__("threading").RLock()
|
||||
manager._fundamental_timeout_worker_limit = 8
|
||||
manager._fundamental_timeout_slots = __import__("threading").BoundedSemaphore(8)
|
||||
manager._run_with_retry = Mock(side_effect=lambda task, timeout, name: (task(), None, 10))
|
||||
return manager
|
||||
|
||||
@patch("data_provider.futu_fetcher.FutuFetcher.has_configured_endpoint", return_value=True)
|
||||
@patch("data_provider.futu_fundamental_adapter.FutuFundamentalAdapter")
|
||||
@patch("src.config.get_config")
|
||||
def test_fetch_offshore_bundle_prefers_futu_for_hk(self, mock_get_config, mock_adapter, mock_has_ep):
|
||||
from types import SimpleNamespace
|
||||
|
||||
mock_get_config.return_value = SimpleNamespace()
|
||||
futu_bundle = {
|
||||
"status": "partial",
|
||||
"growth": {"revenue_yoy": 10.0, "net_profit_yoy": 12.0, "gross_margin": 40.0},
|
||||
"earnings": {
|
||||
"financial_report": {
|
||||
"revenue": 1.0e10,
|
||||
"net_profit_parent": 200.0,
|
||||
"basic_eps": 1.2,
|
||||
"gross_profit": 4.0e9,
|
||||
},
|
||||
"dividend": {
|
||||
"events": [{"event_date": "2026-01-15", "ex_dividend_date": "2026-01-15", "cash_dividend_per_share": 3.5}],
|
||||
"ttm_event_count": 2,
|
||||
"ttm_cash_dividend_per_share": 7.0,
|
||||
"ttm_dividend_yield_pct": 3.2,
|
||||
"source": "futu",
|
||||
},
|
||||
},
|
||||
"institution": {"company_profile": {"公司名称": "测试公司"}},
|
||||
"capital_flow": {"latest": {"main_in_flow": 10.0}},
|
||||
"belong_boards": [{"name": "测试行业", "code": "HK.TEST", "type": "INDUSTRY"}],
|
||||
"source_chain": [{"provider": "futu.financials", "result": "ok", "duration_ms": 1}],
|
||||
"errors": [],
|
||||
}
|
||||
mock_adapter.return_value.get_fundamental_bundle.return_value = futu_bundle
|
||||
manager = self._make_manager()
|
||||
|
||||
payload, err, ms, provider = manager._fetch_offshore_fundamental_bundle("HK00700", "hk", 10.0)
|
||||
|
||||
self.assertIsNone(err)
|
||||
self.assertEqual(provider, "fundamental_bundle_futu")
|
||||
self.assertIs(payload, futu_bundle)
|
||||
# All core growth/earnings fields present -> no field gaps -> no yfinance call.
|
||||
self.assertEqual(manager._yfinance_fundamental_adapter.get_fundamental_bundle.call_count, 0)
|
||||
|
||||
@patch("data_provider.futu_fetcher.FutuFetcher.has_configured_endpoint", return_value=False)
|
||||
@patch("src.config.get_config")
|
||||
def test_fetch_offshore_bundle_uses_yfinance_without_futu(self, mock_get_config, mock_has_ep):
|
||||
from types import SimpleNamespace
|
||||
|
||||
mock_get_config.return_value = SimpleNamespace()
|
||||
manager = self._make_manager()
|
||||
manager._yfinance_fundamental_adapter.get_fundamental_bundle.return_value = {
|
||||
"status": "not_supported", "growth": {}, "earnings": {},
|
||||
"belong_boards": [], "source_chain": [], "errors": [],
|
||||
}
|
||||
|
||||
payload, err, ms, provider = manager._fetch_offshore_fundamental_bundle("HK00700", "hk", 10.0)
|
||||
|
||||
self.assertEqual(provider, "fundamental_bundle_yfinance")
|
||||
self.assertEqual(manager._yfinance_fundamental_adapter.get_fundamental_bundle.call_count, 1)
|
||||
|
||||
@patch("data_provider.futu_fetcher.FutuFetcher.has_configured_endpoint", return_value=True)
|
||||
@patch("data_provider.futu_fundamental_adapter.FutuFundamentalAdapter")
|
||||
@patch("src.config.get_config")
|
||||
def test_fetch_offshore_bundle_falls_back_to_yfinance_when_futu_empty(self, mock_get_config, mock_adapter, mock_has_ep):
|
||||
from types import SimpleNamespace
|
||||
|
||||
mock_get_config.return_value = SimpleNamespace()
|
||||
mock_adapter.return_value.get_fundamental_bundle.return_value = {
|
||||
"status": "not_supported", "growth": {}, "earnings": {},
|
||||
"institution": {}, "capital_flow": {}, "belong_boards": [],
|
||||
"source_chain": [], "errors": [],
|
||||
}
|
||||
manager = self._make_manager()
|
||||
manager._yfinance_fundamental_adapter.get_fundamental_bundle.return_value = {
|
||||
"status": "ok", "growth": {"revenue_yoy": 5.0}, "earnings": {},
|
||||
"belong_boards": [], "source_chain": [], "errors": [],
|
||||
}
|
||||
|
||||
payload, err, ms, provider = manager._fetch_offshore_fundamental_bundle("HK00700", "hk", 10.0)
|
||||
|
||||
self.assertEqual(provider, "fundamental_bundle_yfinance")
|
||||
self.assertEqual(payload["growth"]["revenue_yoy"], 5.0)
|
||||
|
||||
def test_futu_boards_normalize_to_name_code_type_contract(self):
|
||||
"""Futu OpenD plate_* fields must map to DSA's name/type/code contract."""
|
||||
from unittest.mock import Mock
|
||||
|
||||
import pandas as pd
|
||||
|
||||
fetcher = Mock()
|
||||
fetcher.get_owner_plate.return_value = pd.DataFrame(
|
||||
[{"plate_code": "HK.TEST", "plate_name": "测试行业", "plate_type": "INDUSTRY"}]
|
||||
)
|
||||
bundle = FutuFundamentalAdapter(fetcher).get_fundamental_bundle("HK01810")
|
||||
|
||||
boards = bundle["belong_boards"]
|
||||
self.assertEqual(len(boards), 1)
|
||||
self.assertEqual(boards[0]["name"], "测试行业")
|
||||
self.assertEqual(boards[0]["code"], "HK.TEST")
|
||||
self.assertEqual(boards[0]["type"], "INDUSTRY")
|
||||
self.assertNotIn("plate_name", boards[0])
|
||||
self.assertNotIn("plate_code", boards[0])
|
||||
self.assertNotIn("plate_type", boards[0])
|
||||
|
||||
def test_futu_boards_survive_extract_board_detail_fields(self):
|
||||
"""HK Futu belong_boards must be consumable by the board-detail helper."""
|
||||
from src.utils.data_processing import extract_board_detail_fields
|
||||
|
||||
snapshot = {
|
||||
"fundamental_context": {
|
||||
"market": "hk",
|
||||
"belong_boards": [{"name": "测试行业", "code": "HK.TEST", "type": "INDUSTRY"}],
|
||||
},
|
||||
}
|
||||
extracted = extract_board_detail_fields(snapshot)
|
||||
self.assertEqual(extracted["belong_boards"][0]["name"], "测试行业")
|
||||
self.assertEqual(extracted["belong_boards"][0]["code"], "HK.TEST")
|
||||
self.assertEqual(extracted["belong_boards"][0]["type"], "INDUSTRY")
|
||||
|
||||
@patch("data_provider.futu_fetcher.FutuFetcher.has_configured_endpoint", return_value=True)
|
||||
@patch("data_provider.futu_fundamental_adapter.FutuFundamentalAdapter")
|
||||
@patch("src.config.get_config")
|
||||
def test_fetch_offshore_bundle_merges_yfinance_when_futu_missing_growth_earnings(
|
||||
self, mock_get_config, mock_adapter, mock_has_ep
|
||||
):
|
||||
"""Futu partial success must not drop growth/earnings that yfinance still provides."""
|
||||
from types import SimpleNamespace
|
||||
|
||||
mock_get_config.return_value = SimpleNamespace()
|
||||
# Futu only has static info / capital flow / boards; statements failed.
|
||||
mock_adapter.return_value.get_fundamental_bundle.return_value = {
|
||||
"status": "partial",
|
||||
"growth": {},
|
||||
"earnings": {},
|
||||
"institution": {"company_profile": {"公司名称": "测试公司"}},
|
||||
"capital_flow": {"latest": {"main_in_flow": 10.0}},
|
||||
"belong_boards": [{"name": "测试行业", "code": "HK.TEST", "type": "INDUSTRY"}],
|
||||
"source_chain": [{"provider": "futu.financials", "result": "ok", "duration_ms": 1}],
|
||||
"errors": [],
|
||||
}
|
||||
manager = self._make_manager()
|
||||
manager._yfinance_fundamental_adapter.get_fundamental_bundle.return_value = {
|
||||
"status": "ok",
|
||||
"growth": {"revenue_yoy": 16.5, "net_profit_yoy": 19.3},
|
||||
"earnings": {"financial_report": {"net_profit_parent": 2.95e10}},
|
||||
"belong_boards": [],
|
||||
"source_chain": [{"provider": "yfinance.info", "result": "ok", "duration_ms": 1}],
|
||||
"errors": [],
|
||||
}
|
||||
|
||||
payload, err, ms, provider = manager._fetch_offshore_fundamental_bundle("HK00700", "hk", 10.0)
|
||||
|
||||
self.assertEqual(provider, "fundamental_bundle_futu")
|
||||
# yfinance growth/earnings merged in; Futu blocks kept.
|
||||
self.assertEqual(payload["growth"]["revenue_yoy"], 16.5)
|
||||
self.assertEqual(payload["growth"]["net_profit_yoy"], 19.3)
|
||||
self.assertEqual(payload["earnings"]["financial_report"]["net_profit_parent"], 2.95e10)
|
||||
self.assertEqual(payload["institution"]["company_profile"]["公司名称"], "测试公司")
|
||||
self.assertEqual(payload["belong_boards"][0]["name"], "测试行业")
|
||||
self.assertEqual(payload["status"], "partial")
|
||||
providers = [s.get("provider") for s in payload["source_chain"]]
|
||||
self.assertIn("futu.financials", providers)
|
||||
self.assertIn("yfinance.info", providers)
|
||||
|
||||
@patch("data_provider.futu_fetcher.FutuFetcher.has_configured_endpoint", return_value=True)
|
||||
@patch("data_provider.futu_fundamental_adapter.FutuFundamentalAdapter")
|
||||
@patch("src.config.get_config")
|
||||
def test_fetch_offshore_bundle_merges_yfinance_when_futu_growth_is_all_none(
|
||||
self, mock_get_config, mock_adapter, mock_has_ep
|
||||
):
|
||||
"""Truthy but value-less growth/earnings (all-None) must still pull yfinance."""
|
||||
from types import SimpleNamespace
|
||||
|
||||
mock_get_config.return_value = SimpleNamespace()
|
||||
# Futu _financials() writes the fixed key set even when every value is
|
||||
# None or display_name did not match the adapter aliases.
|
||||
mock_adapter.return_value.get_fundamental_bundle.return_value = {
|
||||
"status": "partial",
|
||||
"growth": {"revenue_yoy": None, "net_profit_yoy": None, "gross_margin": None},
|
||||
"earnings": {
|
||||
"financial_report": {"report_date": None, "period": "FY2025", "currency": None},
|
||||
"dividend": {},
|
||||
},
|
||||
"institution": {"company_profile": {"公司名称": "测试公司"}},
|
||||
"capital_flow": {"latest": {"main_in_flow": 10.0}},
|
||||
"belong_boards": [{"name": "测试行业", "code": "HK.TEST", "type": "INDUSTRY"}],
|
||||
"source_chain": [{"provider": "futu.financials", "result": "ok", "duration_ms": 1}],
|
||||
"errors": [],
|
||||
}
|
||||
manager = self._make_manager()
|
||||
manager._yfinance_fundamental_adapter.get_fundamental_bundle.return_value = {
|
||||
"status": "ok",
|
||||
"growth": {"revenue_yoy": 16.5, "net_profit_yoy": 19.3},
|
||||
"earnings": {"financial_report": {"net_profit_parent": 2.95e10}},
|
||||
"belong_boards": [],
|
||||
"source_chain": [{"provider": "yfinance.info", "result": "ok", "duration_ms": 1}],
|
||||
"errors": [],
|
||||
}
|
||||
|
||||
payload, err, ms, provider = manager._fetch_offshore_fundamental_bundle("HK00700", "hk", 10.0)
|
||||
|
||||
self.assertEqual(provider, "fundamental_bundle_futu")
|
||||
# All-None Futu growth replaced by meaningful yfinance values.
|
||||
self.assertEqual(payload["growth"]["revenue_yoy"], 16.5)
|
||||
self.assertEqual(payload["growth"]["net_profit_yoy"], 19.3)
|
||||
self.assertEqual(payload["earnings"]["financial_report"]["net_profit_parent"], 2.95e10)
|
||||
self.assertEqual(payload["institution"]["company_profile"]["公司名称"], "测试公司")
|
||||
providers = [s.get("provider") for s in payload["source_chain"]]
|
||||
self.assertIn("yfinance.info", providers)
|
||||
|
||||
@patch("data_provider.futu_fetcher.FutuFetcher.has_configured_endpoint", return_value=True)
|
||||
@patch("data_provider.futu_fundamental_adapter.FutuFundamentalAdapter")
|
||||
@patch("src.config.get_config")
|
||||
def test_fetch_offshore_bundle_fills_partial_futu_fields_from_yfinance(
|
||||
self, mock_get_config, mock_adapter, mock_has_ep
|
||||
):
|
||||
"""Partial Futu hits (some core fields present) must not drop the rest."""
|
||||
from types import SimpleNamespace
|
||||
|
||||
mock_get_config.return_value = SimpleNamespace()
|
||||
# Futu matched only some aliases: growth has revenue_yoy but net_profit_yoy
|
||||
# / gross_margin are None; earnings only has basic_eps.
|
||||
mock_adapter.return_value.get_fundamental_bundle.return_value = {
|
||||
"status": "partial",
|
||||
"growth": {"revenue_yoy": 10.0, "net_profit_yoy": None, "gross_margin": 40.0},
|
||||
"earnings": {"financial_report": {"basic_eps": 0.2}},
|
||||
"institution": {"company_profile": {"公司名称": "测试公司"}},
|
||||
"capital_flow": {},
|
||||
"belong_boards": [],
|
||||
"source_chain": [{"provider": "futu.financials", "result": "ok", "duration_ms": 1}],
|
||||
"errors": [],
|
||||
}
|
||||
manager = self._make_manager()
|
||||
manager._yfinance_fundamental_adapter.get_fundamental_bundle.return_value = {
|
||||
"status": "ok",
|
||||
"growth": {"revenue_yoy": 16.5, "net_profit_yoy": 19.3, "gross_margin": 47.8},
|
||||
"earnings": {
|
||||
"financial_report": {
|
||||
"revenue": 1.11e11,
|
||||
"net_profit_parent": 2.95e10,
|
||||
"basic_eps": 1.9,
|
||||
}
|
||||
},
|
||||
"belong_boards": [],
|
||||
"source_chain": [{"provider": "yfinance.info", "result": "ok", "duration_ms": 1}],
|
||||
"errors": [],
|
||||
}
|
||||
|
||||
payload, err, ms, provider = manager._fetch_offshore_fundamental_bundle("HK00700", "hk", 10.0)
|
||||
|
||||
self.assertEqual(provider, "fundamental_bundle_futu")
|
||||
self.assertEqual(manager._yfinance_fundamental_adapter.get_fundamental_bundle.call_count, 1)
|
||||
# Futu-present fields stay; missing fields filled from yfinance.
|
||||
self.assertEqual(payload["growth"]["revenue_yoy"], 10.0)
|
||||
self.assertEqual(payload["growth"]["net_profit_yoy"], 19.3)
|
||||
self.assertEqual(payload["growth"]["gross_margin"], 40.0)
|
||||
self.assertEqual(payload["earnings"]["financial_report"]["basic_eps"], 0.2)
|
||||
self.assertEqual(payload["earnings"]["financial_report"]["revenue"], 1.11e11)
|
||||
self.assertEqual(payload["earnings"]["financial_report"]["net_profit_parent"], 2.95e10)
|
||||
providers = [s.get("provider") for s in payload["source_chain"]]
|
||||
self.assertIn("futu.financials", providers)
|
||||
self.assertIn("yfinance.info", providers)
|
||||
|
||||
@patch("data_provider.futu_fetcher.FutuFetcher.has_configured_endpoint", return_value=True)
|
||||
@patch("data_provider.futu_fundamental_adapter.FutuFundamentalAdapter")
|
||||
@patch("src.config.get_config")
|
||||
def test_fetch_offshore_bundle_pulls_yfinance_when_futu_dividend_empty(
|
||||
self, mock_get_config, mock_adapter, mock_has_ep
|
||||
):
|
||||
"""Complete Futu growth/financial_report but empty dividend must still pull yfinance."""
|
||||
from types import SimpleNamespace
|
||||
|
||||
mock_get_config.return_value = SimpleNamespace()
|
||||
mock_adapter.return_value.get_fundamental_bundle.return_value = {
|
||||
"status": "partial",
|
||||
"growth": {"revenue_yoy": 10.0, "net_profit_yoy": 12.0, "gross_margin": 40.0},
|
||||
"earnings": {
|
||||
"financial_report": {
|
||||
"revenue": 1.0e10,
|
||||
"net_profit_parent": 200.0,
|
||||
"basic_eps": 1.2,
|
||||
"gross_profit": 4.0e9,
|
||||
},
|
||||
"dividend": {"events": [], "source": "futu"},
|
||||
},
|
||||
"institution": {"company_profile": {"公司名称": "测试公司"}},
|
||||
"capital_flow": {"latest": {"main_in_flow": 10.0}},
|
||||
"belong_boards": [],
|
||||
"source_chain": [{"provider": "futu.financials", "result": "ok", "duration_ms": 1}],
|
||||
"errors": [],
|
||||
}
|
||||
manager = self._make_manager()
|
||||
manager._yfinance_fundamental_adapter.get_fundamental_bundle.return_value = {
|
||||
"status": "ok",
|
||||
"growth": {"revenue_yoy": 16.5},
|
||||
"earnings": {
|
||||
"financial_report": {"revenue": 1.11e11},
|
||||
"dividend": {
|
||||
"events": [{"event_date": "2026-01-15", "ex_dividend_date": "2026-01-15", "cash_dividend_per_share": 3.5}],
|
||||
"ttm_event_count": 2,
|
||||
"ttm_cash_dividend_per_share": 7.0,
|
||||
"ttm_dividend_yield_pct": 3.2,
|
||||
},
|
||||
},
|
||||
"belong_boards": [],
|
||||
"source_chain": [{"provider": "yfinance.info", "result": "ok", "duration_ms": 1}],
|
||||
"errors": [],
|
||||
}
|
||||
|
||||
payload, err, ms, provider = manager._fetch_offshore_fundamental_bundle("HK00700", "hk", 10.0)
|
||||
|
||||
self.assertEqual(provider, "fundamental_bundle_futu")
|
||||
self.assertEqual(manager._yfinance_fundamental_adapter.get_fundamental_bundle.call_count, 1)
|
||||
# Futu growth/earnings kept, dividend contract filled from yfinance.
|
||||
self.assertEqual(payload["growth"]["revenue_yoy"], 10.0)
|
||||
self.assertEqual(payload["earnings"]["financial_report"]["net_profit_parent"], 200.0)
|
||||
self.assertEqual(payload["earnings"]["dividend"]["ttm_cash_dividend_per_share"], 7.0)
|
||||
self.assertEqual(payload["earnings"]["dividend"]["ttm_dividend_yield_pct"], 3.2)
|
||||
self.assertEqual(payload["earnings"]["dividend"]["events"][0]["cash_dividend_per_share"], 3.5)
|
||||
providers = [s.get("provider") for s in payload["source_chain"]]
|
||||
self.assertIn("yfinance.info", providers)
|
||||
|
||||
def test_dividends_normalize_opend_fields_to_repo_contract(self):
|
||||
"""OpenD raw dividend fields (ex_date/record_date/statement) must map to the repo contract."""
|
||||
from unittest.mock import Mock
|
||||
|
||||
fetcher = Mock()
|
||||
fetcher.get_financials_statements.return_value = {"report_list": []}
|
||||
fetcher.get_stock_basicinfo.return_value = {
|
||||
"static_info": {"lot_size": 200, "suspension": False},
|
||||
"company_profile": {"公司名称": "测试公司"},
|
||||
}
|
||||
fetcher.get_corporate_actions_dividends.return_value = {
|
||||
"dividend_list": [
|
||||
{
|
||||
"ex_date": "2026-06-30",
|
||||
"record_date": "2026-07-02",
|
||||
"statement": "FY2025",
|
||||
"dividend_per_share": 1.25,
|
||||
"currency": "HKD",
|
||||
"description": "Final dividend",
|
||||
},
|
||||
{
|
||||
"ex_date": "2026-01-15",
|
||||
"record_date": "2026-01-16",
|
||||
"statement": "FY2024",
|
||||
"dividend_per_share": 1.1,
|
||||
"currency": "HKD",
|
||||
},
|
||||
]
|
||||
}
|
||||
fetcher.get_corporate_actions_stock_splits.return_value = {"split_list": []}
|
||||
fetcher.get_capital_flow.return_value = None
|
||||
fetcher.get_owner_plate.return_value = None
|
||||
fetcher.get_realtime_quote.return_value = {"price": 50.0}
|
||||
|
||||
bundle = FutuFundamentalAdapter(fetcher).get_fundamental_bundle("HK00700")
|
||||
|
||||
dividend = bundle["earnings"]["dividend"]
|
||||
self.assertEqual(dividend["source"], "futu")
|
||||
self.assertEqual(dividend["events"][0]["event_date"], "2026-06-30")
|
||||
self.assertEqual(dividend["events"][0]["ex_dividend_date"], "2026-06-30")
|
||||
self.assertEqual(dividend["events"][0]["record_date"], "2026-07-02")
|
||||
self.assertEqual(dividend["events"][0]["cash_dividend_per_share"], 1.25)
|
||||
self.assertEqual(dividend["events"][0]["statement"], "FY2025")
|
||||
# TTM cash = 1.25 + 1.1, yield = ttm / price.
|
||||
self.assertEqual(dividend["ttm_event_count"], 2)
|
||||
self.assertEqual(dividend["ttm_cash_dividend_per_share"], 2.35)
|
||||
self.assertEqual(dividend["ttm_dividend_yield_pct"], round(2.35 / 50.0 * 100.0, 4))
|
||||
self.assertNotIn("ex_date", dividend["events"][0])
|
||||
|
||||
def test_dividends_compute_yield_from_unified_quote_object(self):
|
||||
"""Yield must be computed when get_realtime_quote returns a UnifiedRealtimeQuote dataclass."""
|
||||
from unittest.mock import Mock
|
||||
|
||||
from data_provider.realtime_types import RealtimeSource, UnifiedRealtimeQuote
|
||||
|
||||
fetcher = Mock()
|
||||
fetcher.get_financials_statements.return_value = {"report_list": []}
|
||||
fetcher.get_stock_basicinfo.return_value = {
|
||||
"static_info": {"lot_size": 200, "suspension": False},
|
||||
"company_profile": {"公司名称": "测试公司"},
|
||||
}
|
||||
fetcher.get_corporate_actions_dividends.return_value = {
|
||||
"dividend_list": [
|
||||
{"ex_date": "2026-06-30", "dividend_per_share": 1.25, "currency": "HKD"},
|
||||
]
|
||||
}
|
||||
fetcher.get_corporate_actions_stock_splits.return_value = {"split_list": []}
|
||||
fetcher.get_capital_flow.return_value = None
|
||||
fetcher.get_owner_plate.return_value = None
|
||||
# Real FutuFetcher shape: UnifiedRealtimeQuote, not dict.
|
||||
fetcher.get_realtime_quote.return_value = UnifiedRealtimeQuote(
|
||||
code="HK00700", name="Tencent", price=50.0, source=RealtimeSource.FUTU
|
||||
)
|
||||
|
||||
bundle = FutuFundamentalAdapter(fetcher).get_fundamental_bundle("HK00700")
|
||||
|
||||
dividend = bundle["earnings"]["dividend"]
|
||||
self.assertEqual(dividend["ttm_cash_dividend_per_share"], 1.25)
|
||||
self.assertEqual(dividend["ttm_dividend_yield_pct"], round(1.25 / 50.0 * 100.0, 4))
|
||||
|
||||
def test_dividends_missing_yield_when_quote_unavailable(self):
|
||||
"""A failed/no-price realtime quote must leave a contract gap, not a complete block.
|
||||
|
||||
The dividend block keeps events and TTM cash, but without a price the
|
||||
repo contract field ttm_dividend_yield_pct cannot be computed. This is
|
||||
the exact shape the manager's supplement logic must detect as a gap.
|
||||
"""
|
||||
from unittest.mock import Mock
|
||||
|
||||
fetcher = Mock()
|
||||
fetcher.get_financials_statements.return_value = {"report_list": []}
|
||||
fetcher.get_stock_basicinfo.return_value = {
|
||||
"static_info": {"lot_size": 200, "suspension": False},
|
||||
"company_profile": {"公司名称": "测试公司"},
|
||||
}
|
||||
fetcher.get_corporate_actions_dividends.return_value = {
|
||||
"dividend_list": [
|
||||
{"ex_date": "2026-06-30", "dividend_per_share": 1.25, "currency": "HKD"},
|
||||
]
|
||||
}
|
||||
fetcher.get_corporate_actions_stock_splits.return_value = {"split_list": []}
|
||||
fetcher.get_capital_flow.return_value = None
|
||||
fetcher.get_owner_plate.return_value = None
|
||||
# OpenD quote snapshot unavailable / no price -> yield cannot be computed.
|
||||
fetcher.get_realtime_quote.return_value = None
|
||||
|
||||
bundle = FutuFundamentalAdapter(fetcher).get_fundamental_bundle("HK00700")
|
||||
|
||||
dividend = bundle["earnings"]["dividend"]
|
||||
self.assertEqual(dividend["ttm_cash_dividend_per_share"], 1.25)
|
||||
self.assertNotIn("ttm_dividend_yield_pct", dividend)
|
||||
self.assertEqual(dividend["events"][0]["cash_dividend_per_share"], 1.25)
|
||||
|
||||
@patch("data_provider.futu_fetcher.FutuFetcher.has_configured_endpoint", return_value=True)
|
||||
@patch("data_provider.futu_fundamental_adapter.FutuFundamentalAdapter")
|
||||
@patch("src.config.get_config")
|
||||
def test_fetch_offshore_bundle_pulls_yfinance_when_futu_dividend_missing_yield(
|
||||
self, mock_get_config, mock_adapter, mock_has_ep
|
||||
):
|
||||
"""Futu dividend with TTM cash but no yield (quote price unavailable) must pull yfinance.
|
||||
|
||||
The repo contract consumes ttm_cash_dividend_per_share and
|
||||
ttm_dividend_yield_pct together. When Futu keeps events + TTM cash but
|
||||
the extra realtime quote failed (no price -> no yield), the block must
|
||||
still count as a gap so yfinance can fill the missing yield instead of
|
||||
the notification rendering N/A.
|
||||
"""
|
||||
from types import SimpleNamespace
|
||||
|
||||
mock_get_config.return_value = SimpleNamespace()
|
||||
mock_adapter.return_value.get_fundamental_bundle.return_value = {
|
||||
"status": "partial",
|
||||
"growth": {"revenue_yoy": 10.0, "net_profit_yoy": 12.0, "gross_margin": 40.0},
|
||||
"earnings": {
|
||||
"financial_report": {
|
||||
"revenue": 1.0e10,
|
||||
"net_profit_parent": 200.0,
|
||||
"basic_eps": 1.2,
|
||||
"gross_profit": 4.0e9,
|
||||
},
|
||||
# Realistic shape after get_realtime_quote() returned None/0:
|
||||
# events + TTM cash are present, but yield is missing.
|
||||
"dividend": {
|
||||
"events": [
|
||||
{
|
||||
"event_date": "2026-01-15",
|
||||
"ex_dividend_date": "2026-01-15",
|
||||
"cash_dividend_per_share": 3.5,
|
||||
}
|
||||
],
|
||||
"ttm_event_count": 2,
|
||||
"ttm_cash_dividend_per_share": 7.0,
|
||||
"source": "futu",
|
||||
},
|
||||
},
|
||||
"institution": {"company_profile": {"公司名称": "测试公司"}},
|
||||
"capital_flow": {"latest": {"main_in_flow": 10.0}},
|
||||
"belong_boards": [],
|
||||
"source_chain": [{"provider": "futu.financials", "result": "ok", "duration_ms": 1}],
|
||||
"errors": [],
|
||||
}
|
||||
manager = self._make_manager()
|
||||
manager._yfinance_fundamental_adapter.get_fundamental_bundle.return_value = {
|
||||
"status": "ok",
|
||||
"growth": {"revenue_yoy": 16.5},
|
||||
"earnings": {
|
||||
"financial_report": {"revenue": 1.11e11},
|
||||
"dividend": {
|
||||
"events": [
|
||||
{
|
||||
"event_date": "2026-01-15",
|
||||
"ex_dividend_date": "2026-01-15",
|
||||
"cash_dividend_per_share": 3.5,
|
||||
}
|
||||
],
|
||||
"ttm_event_count": 2,
|
||||
"ttm_cash_dividend_per_share": 7.0,
|
||||
"ttm_dividend_yield_pct": 3.2,
|
||||
},
|
||||
},
|
||||
"belong_boards": [],
|
||||
"source_chain": [{"provider": "yfinance.info", "result": "ok", "duration_ms": 1}],
|
||||
"errors": [],
|
||||
}
|
||||
|
||||
payload, err, ms, provider = manager._fetch_offshore_fundamental_bundle("HK00700", "hk", 10.0)
|
||||
|
||||
self.assertEqual(provider, "fundamental_bundle_futu")
|
||||
self.assertEqual(manager._yfinance_fundamental_adapter.get_fundamental_bundle.call_count, 1)
|
||||
# Futu growth/earnings kept; the missing yield is filled from yfinance.
|
||||
self.assertEqual(payload["earnings"]["dividend"]["ttm_cash_dividend_per_share"], 7.0)
|
||||
self.assertEqual(payload["earnings"]["dividend"]["ttm_dividend_yield_pct"], 3.2)
|
||||
providers = [s.get("provider") for s in payload["source_chain"]]
|
||||
self.assertIn("yfinance.info", providers)
|
||||
|
||||
def test_close_releases_futu_fundamental_fetcher(self):
|
||||
"""DataFetcherManager.close() must close the cached HK Futu fundamental fetcher.
|
||||
|
||||
The HK Futu fundamental path lazily caches its own FutuFetcher (an
|
||||
OpenQuoteContext-backed connection) on _futu_fundamental_fetcher.
|
||||
Explicit close / reload paths must release it, otherwise the OpenD
|
||||
connection stays open after manager cleanup.
|
||||
"""
|
||||
from unittest.mock import Mock
|
||||
|
||||
manager = self._make_manager()
|
||||
futu_fetcher = Mock()
|
||||
manager._futu_fundamental_fetcher = futu_fetcher
|
||||
|
||||
manager.close()
|
||||
|
||||
futu_fetcher.close.assert_called_once_with()
|
||||
self.assertIsNone(manager._futu_fundamental_fetcher)
|
||||
@@ -50,6 +50,130 @@ class TestHKRealtimeRouting(unittest.TestCase):
|
||||
self.assertEqual(efinance.calls, [])
|
||||
self.assertEqual(tushare.calls, [])
|
||||
|
||||
@patch("data_provider.futu_fetcher.FutuFetcher.has_configured_endpoint", return_value=True)
|
||||
@patch("src.config.get_config")
|
||||
def test_manager_routes_hk_through_configured_futu_priority(self, mock_get_config, mock_has_ep):
|
||||
mock_get_config.return_value = SimpleNamespace(
|
||||
enable_realtime_quote=True,
|
||||
realtime_source_priority="tencent,akshare_sina,efinance,akshare_em,tushare",
|
||||
futu_hk_realtime_source_priority="futu,akshare,yfinance",
|
||||
)
|
||||
futu_quote = MagicMock()
|
||||
futu_quote.has_basic_data.return_value = True
|
||||
futu = _DummyFetcher("FutuFetcher", 0, result=futu_quote)
|
||||
akshare = _DummyFetcher("AkshareFetcher", 1, result=None)
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
manager = DataFetcherManager(fetchers=[futu, akshare])
|
||||
quote = manager.get_realtime_quote("HK01810")
|
||||
|
||||
self.assertIs(quote, futu_quote)
|
||||
self.assertEqual(len(futu.calls), 1)
|
||||
self.assertEqual(akshare.calls, [])
|
||||
|
||||
@patch("data_provider.futu_fetcher.FutuFetcher.has_configured_endpoint", return_value=True)
|
||||
@patch("src.config.get_config")
|
||||
def test_manager_falls_back_from_futu_to_akshare(self, mock_get_config, mock_has_ep):
|
||||
mock_get_config.return_value = SimpleNamespace(
|
||||
enable_realtime_quote=True,
|
||||
realtime_source_priority="tencent,akshare_sina,efinance,akshare_em,tushare",
|
||||
futu_hk_realtime_source_priority="futu,akshare,yfinance",
|
||||
)
|
||||
futu = _DummyFetcher("FutuFetcher", 0, result=None)
|
||||
akshare_quote = MagicMock()
|
||||
akshare_quote.has_basic_data.return_value = True
|
||||
akshare = _DummyFetcher("AkshareFetcher", 1, result=akshare_quote)
|
||||
|
||||
manager = DataFetcherManager(fetchers=[futu, akshare])
|
||||
quote = manager.get_realtime_quote("HK01810")
|
||||
|
||||
self.assertIs(quote, akshare_quote)
|
||||
self.assertEqual(len(futu.calls), 1)
|
||||
self.assertEqual(akshare.calls, [((("HK01810",), {"source": "hk"}))])
|
||||
# 首选源 Futu 失败、次源 AkShare 接管时，应保留 fallback_from 元数据。
|
||||
self.assertEqual(getattr(quote, "fallback_from", None), "futu")
|
||||
|
||||
@patch("data_provider.futu_fetcher.FutuFetcher.has_configured_endpoint", return_value=False)
|
||||
@patch("src.config.get_config")
|
||||
def test_manager_skips_unconfigured_futu_without_fallback_from(self, mock_get_config, mock_has_ep):
|
||||
"""An unconfigured Futu source must be skipped, not recorded as the failed primary.
|
||||
|
||||
With FUTU_OPEND_HOST unset, the default HK priority
|
||||
(futu,longbridge,akshare,yfinance) must not treat the never-enabled
|
||||
futu source as a failed primary: the first successfully enabled
|
||||
source's quote should carry no fallback_from at all.
|
||||
"""
|
||||
mock_get_config.return_value = SimpleNamespace(
|
||||
enable_realtime_quote=True,
|
||||
realtime_source_priority="tencent,akshare_sina,efinance,akshare_em,tushare",
|
||||
futu_hk_realtime_source_priority="futu,akshare,yfinance",
|
||||
)
|
||||
futu = _DummyFetcher("FutuFetcher", 0, result=None)
|
||||
akshare_quote = MagicMock()
|
||||
akshare_quote.has_basic_data.return_value = True
|
||||
akshare = _DummyFetcher("AkshareFetcher", 1, result=akshare_quote)
|
||||
|
||||
manager = DataFetcherManager(fetchers=[futu, akshare])
|
||||
enrich = MagicMock(return_value=akshare_quote)
|
||||
manager._enrich_realtime_quote = enrich
|
||||
quote = manager.get_realtime_quote("HK01810")
|
||||
|
||||
self.assertIs(quote, akshare_quote)
|
||||
# futu is skipped entirely: never called, never recorded as fallback.
|
||||
self.assertEqual(futu.calls, [])
|
||||
self.assertEqual(akshare.calls, [((("HK01810",), {"source": "hk"}))])
|
||||
self.assertEqual(enrich.call_args.kwargs.get("fallback_from"), None)
|
||||
|
||||
@patch("data_provider.futu_fetcher.FutuFetcher.has_configured_endpoint", return_value=True)
|
||||
@patch("src.config.get_config")
|
||||
def test_manager_supplements_partial_futu_quote_from_akshare(self, mock_get_config, mock_has_ep):
|
||||
"""A partial first-source quote should be supplemented by later sources."""
|
||||
mock_get_config.return_value = SimpleNamespace(
|
||||
enable_realtime_quote=True,
|
||||
realtime_source_priority="tencent,akshare_sina,efinance,akshare_em,tushare",
|
||||
futu_hk_realtime_source_priority="futu,akshare,yfinance",
|
||||
realtime_cache_ttl=None,
|
||||
)
|
||||
futu_quote = MagicMock()
|
||||
futu_quote.has_basic_data.return_value = True
|
||||
for field in DataFetcherManager._SUPPLEMENT_FIELDS:
|
||||
setattr(futu_quote, field, None)
|
||||
futu = _DummyFetcher("FutuFetcher", 0, result=futu_quote)
|
||||
|
||||
akshare_quote = MagicMock()
|
||||
akshare_quote.has_basic_data.return_value = True
|
||||
for field in DataFetcherManager._SUPPLEMENT_FIELDS:
|
||||
setattr(akshare_quote, field, 1.86)
|
||||
akshare = _DummyFetcher("AkshareFetcher", 1, result=akshare_quote)
|
||||
|
||||
manager = DataFetcherManager(fetchers=[futu, akshare])
|
||||
quote = manager.get_realtime_quote("HK00700")
|
||||
|
||||
self.assertIs(quote, futu_quote)
|
||||
for field in DataFetcherManager._SUPPLEMENT_FIELDS:
|
||||
self.assertEqual(getattr(quote, field), 1.86, field)
|
||||
self.assertEqual(len(futu.calls), 1)
|
||||
self.assertEqual(akshare.calls, [(("HK00700",), {"source": "hk"})])
|
||||
|
||||
@patch("data_provider.futu_fetcher.FutuFetcher.has_configured_endpoint", return_value=True)
|
||||
@patch("src.config.get_config")
|
||||
def test_manager_does_not_supplement_when_primary_is_complete(self, mock_get_config, mock_has_ep):
|
||||
"""A complete first-source quote should not trigger extra source calls."""
|
||||
mock_get_config.return_value = SimpleNamespace(
|
||||
enable_realtime_quote=True,
|
||||
realtime_source_priority="tencent,akshare_sina,efinance,akshare_em,tushare",
|
||||
futu_hk_realtime_source_priority="futu,akshare,yfinance",
|
||||
realtime_cache_ttl=None,
|
||||
)
|
||||
futu_quote = MagicMock()
|
||||
futu_quote.has_basic_data.return_value = True
|
||||
for field in DataFetcherManager._SUPPLEMENT_FIELDS:
|
||||
setattr(futu_quote, field, 1.0)
|
||||
futu = _DummyFetcher("FutuFetcher", 0, result=futu_quote)
|
||||
akshare = _DummyFetcher("AkshareFetcher", 1, result=None)
|
||||
|
||||
manager = DataFetcherManager(fetchers=[futu, akshare])
|
||||
quote = manager.get_realtime_quote("HK00700")
|
||||
|
||||
self.assertIs(quote, futu_quote)
|
||||
self.assertEqual(len(futu.calls), 1)
|
||||
self.assertEqual(akshare.calls, [])
|
||||
|
||||
Reference in New Issue
Block a user