from __future__ import annotations
|
|
import time
|
from concurrent.futures import Future, ThreadPoolExecutor, wait
|
from typing import Any, Callable
|
|
from .cache import BlobIntegrityError
|
from .http_client import HttpClient
|
from .providers import acquire_announcements, acquire_finance, acquire_forecast, acquire_market
|
|
|
def acquire_all(
|
client: HttpClient,
|
ticker: str,
|
as_of: str,
|
baseline_results: dict[str, dict[str, Any]] | None = None,
|
) -> dict[str, dict[str, Any]]:
|
jobs: dict[str, Callable[[HttpClient, str, str], dict[str, Any]]] = {
|
"announcements": acquire_announcements,
|
"market": acquire_market,
|
"finance": acquire_finance,
|
"forecast": acquire_forecast,
|
}
|
baseline_results = baseline_results or {}
|
executor = ThreadPoolExecutor(max_workers=4, thread_name_prefix="valuation-v2")
|
futures: dict[Future[dict[str, Any]], str] = {
|
executor.submit(func, client, ticker, as_of): name
|
for name, func in jobs.items()
|
if name not in baseline_results
|
}
|
results: dict[str, dict[str, Any]] = dict(baseline_results)
|
try:
|
remaining = client.remaining()
|
done, pending = wait(futures, timeout=remaining)
|
for future in done:
|
name = futures[future]
|
try:
|
results[name] = future.result()
|
except BaseException as exc:
|
blocking = name != "forecast"
|
results[name] = {
|
"provider_id": name,
|
"adapter_version": "1.0.0",
|
"status": "BLOCKED" if blocking else "GAP",
|
"fetched_at": time.time(),
|
"as_of_date": as_of,
|
"records": [],
|
"sources": [],
|
"gaps": [
|
{
|
"gap_id": f"E_{name.upper()}_FAILED" if blocking else "W_FORECAST_FAILED",
|
"provider": name,
|
"field": name,
|
"reason": f"{type(exc).__name__}: {exc}",
|
"impact": "核心数据不可用" if blocking else "机构预测不可用",
|
"blocking": blocking,
|
"budget_used_seconds": None,
|
"manual_action": "核对登记来源或稍后重试",
|
}
|
],
|
"warnings": [],
|
"raw_artifact_hashes": [],
|
"request_telemetry": [],
|
"cache_integrity_failure": isinstance(exc, BlobIntegrityError),
|
}
|
if pending and hasattr(client, "cancel"):
|
client.cancel()
|
finished_after_cancel, still_pending = wait(pending, timeout=1.0)
|
pending = still_pending
|
for future in finished_after_cancel:
|
name = futures[future]
|
try:
|
results[name] = future.result()
|
except BaseException:
|
pending.add(future)
|
for future in pending:
|
future.cancel()
|
name = futures[future]
|
blocking = name != "forecast"
|
results[name] = {
|
"provider_id": name,
|
"adapter_version": "1.0.0",
|
"status": "BLOCKED" if blocking else "GAP",
|
"fetched_at": time.time(),
|
"as_of_date": as_of,
|
"records": [],
|
"sources": [],
|
"gaps": [
|
{
|
"gap_id": "E_NETWORK_DEADLINE" if blocking else "W_NETWORK_DEADLINE",
|
"provider": name,
|
"field": name,
|
"reason": "90 秒网络截止已到",
|
"impact": "核心数据不可用" if blocking else "机构预测不可用",
|
"blocking": blocking,
|
"budget_used_seconds": 90.0,
|
"manual_action": "稍后重试",
|
}
|
],
|
"warnings": [],
|
"raw_artifact_hashes": [],
|
"request_telemetry": [],
|
"cache_integrity_failure": False,
|
}
|
finally:
|
executor.shutdown(wait=False, cancel_futures=True)
|
kinds = {
|
"announcements": {"stock_identity", "announcement_index"},
|
"market": {"market_close", "shares_market_cap"},
|
"finance": {"finance_main", "finance_income", "finance_balance", "finance_cashflow"},
|
"forecast": {"forecast_summary", "forecast_detail"},
|
}
|
telemetry = getattr(client, "telemetry", [])
|
for name, result in results.items():
|
data_kinds = set(result.get("data_kinds", {}))
|
baseline_kinds = set(result.get("baseline_data_kinds", []))
|
result["baseline_reused"] = bool(data_kinds) and data_kinds <= baseline_kinds
|
if result.get("status") not in {"BLOCKED", "GAP"}:
|
continue
|
partial = [item for item in telemetry if item.get("data_kind") in kinds[name]]
|
result["request_telemetry"] = partial
|
result["raw_artifact_hashes"] = sorted(
|
{item["raw_hash"] for item in partial if item.get("raw_hash")}
|
)
|
return results
|
