stock-data-api / app /api /routes.py
fromozuzhouzzz
feat: add niuone US market + X timeline sources on HF tip
b2ca3f2
Raw
History Blame Contribute Delete
28.1 kB
"""HTTP routes for the stock data API."""
from __future__ import annotations
import time
from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel, Field
from app.core.cache import cache
from app.core.security import require_api_key
from app.services.market_data import market_data_service
from app.services.ai_search_hub_service import AISearchHubNotConfigured, ai_search_hub_service
from app.services.search_service import SearchConfigMissing, SearchUpstreamFailed, search_service
from app.services.source_runner import AllSourcesFailed, get_source_pool_stats
from app.utils.stock_code import InvalidSymbolError
router = APIRouter(dependencies=[Depends(require_api_key)])
GET_SEARCH_SYSTEM_PROMPT = (
"You are a concise financial search assistant. Answer in the user's language. "
"Prioritize verifiable facts, key risks, and fresh market context. Keep the answer short."
)
GET_SEARCH_DEFAULT_ATTEMPTS = 1
GET_SEARCH_DEFAULT_TIMEOUT_SECONDS = 60
GET_SEARCH_DEFAULT_MAX_TOKENS = 1024
GET_SEARCH_FAST_SITES = ["qwen"]
class SearchRequest(BaseModel):
query: str = Field(..., min_length=1, max_length=8000)
context: str | None = Field(default=None, max_length=16000)
system_prompt: str | None = Field(default=None, max_length=4000)
max_attempts: int | None = Field(default=None, ge=1, le=10)
timeout_seconds: int | None = Field(default=None, ge=30, le=600)
temperature: float | None = Field(default=None, ge=0, le=2)
max_tokens: int | None = Field(default=None, ge=256, le=16000)
use_cache: bool = True
class AISearchHubRequest(BaseModel):
query: str = Field(..., min_length=1, max_length=8000)
sites: list[str] | None = Field(default=None, description="Platform keys to search (e.g. gemini, grok, doubao)")
timeout: int | None = Field(default=None, ge=30, le=300)
headless: bool | None = Field(default=None)
use_cache: bool = True
def _handle(func):
try:
return func()
except AllSourcesFailed as exc:
raise HTTPException(
status_code=503,
detail={
"ok": False,
"code": "upstream_unavailable",
"message": "all upstream data sources failed",
"endpoint": exc.endpoint,
"attempts": exc.attempts,
},
) from exc
except InvalidSymbolError as exc:
raise HTTPException(
status_code=400,
detail={
"ok": False,
"code": exc.code,
"message": str(exc),
"symbol": exc.symbol,
},
) from exc
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
except SearchConfigMissing as exc:
raise HTTPException(
status_code=503,
detail={
"ok": False,
"code": "search_config_missing",
"message": f"search service is not configured: {exc}",
"missing_config": exc.missing_config,
},
) from exc
except SearchUpstreamFailed as exc:
raise HTTPException(
status_code=502,
detail={
"message": "upstream search service failed",
"error": str(exc),
"attempts": exc.attempts,
},
) from exc
except AISearchHubNotConfigured as exc:
raise HTTPException(
status_code=503,
detail={
"ok": False,
"code": "ai_search_hub_not_configured",
"message": str(exc),
"missing_config": exc.missing_config,
},
) from exc
except RuntimeError as exc:
raise HTTPException(
status_code=502,
detail={
"message": "upstream search service failed",
"error": str(exc),
},
) from exc
def _search_get_response(
q: str,
max_attempts: int | None,
timeout_seconds: int | None,
temperature: float | None,
max_tokens: int | None,
use_cache: bool,
):
attempts = max_attempts if max_attempts is not None else GET_SEARCH_DEFAULT_ATTEMPTS
timeout = timeout_seconds if timeout_seconds is not None else GET_SEARCH_DEFAULT_TIMEOUT_SECONDS
token_limit = max_tokens if max_tokens is not None else GET_SEARCH_DEFAULT_MAX_TOKENS
use_model_first = max_attempts is not None or temperature is not None or max_tokens is not None
if not use_model_first:
fallback_error: Exception | None = None
fallback_started = time.perf_counter()
try:
payload = ai_search_hub_service.search(
query=q,
sites=GET_SEARCH_FAST_SITES,
timeout=min(max(timeout, 30), 90),
use_cache=use_cache,
)
data = payload.get("data") if isinstance(payload, dict) else None
if isinstance(data, dict) and not data.get("sites_succeeded"):
raise RuntimeError("ai search hub returned no successful sites")
elapsed_ms = int((time.perf_counter() - fallback_started) * 1000)
meta = payload.setdefault("meta", {})
meta["endpoint"] = "search"
meta["source"] = "ai_search_hub.multi_platform.fast_get"
meta["attempts"] = [
{
"source": "ai_search_hub.multi_platform",
"ok": True,
"attempt": 1,
"elapsed_ms": elapsed_ms,
}
]
meta["routing"] = {
"mode": "fast_get",
"primary": "ai_search_hub.multi_platform",
"model_search_available_with": "max_attempts, timeout_seconds, temperature, or max_tokens",
}
return payload
except (AISearchHubNotConfigured, RuntimeError, ValueError) as exc:
fallback_error = exc
try:
payload = search_service.search(
query=q,
system_prompt=GET_SEARCH_SYSTEM_PROMPT,
max_attempts=attempts,
timeout_seconds=timeout,
temperature=temperature,
max_tokens=token_limit,
use_cache=use_cache,
)
if not use_model_first and fallback_error is not None:
meta = payload.setdefault("meta", {})
meta["fallback"] = {
"from": "ai_search_hub.multi_platform",
"to": "openai_compatible.search_model",
"reason": type(fallback_error).__name__,
}
return payload
except (SearchConfigMissing, SearchUpstreamFailed) as exc:
fallback_started = time.perf_counter()
payload = ai_search_hub_service.search(
query=q,
timeout=min(max(timeout, 30), 90),
use_cache=use_cache,
)
elapsed_ms = int((time.perf_counter() - fallback_started) * 1000)
meta = payload.setdefault("meta", {})
primary_attempts = getattr(exc, "attempts", [])
meta["endpoint"] = "search"
meta["source"] = "ai_search_hub.multi_platform.fallback"
meta["attempts"] = [
{"source": "openai_compatible.search_model", **attempt}
for attempt in primary_attempts
] + [
{
"source": "ai_search_hub.multi_platform",
"ok": True,
"attempt": 1,
"elapsed_ms": elapsed_ms,
}
]
meta["fallback"] = {
"from": "openai_compatible.search_model",
"to": "ai_search_hub.multi_platform",
"reason": type(exc).__name__,
}
return payload
@router.get("/catalog")
def catalog():
return market_data_service.catalog()
@router.get("/cache/stats")
def cache_stats():
return cache.stats()
@router.post("/cache/purge-expired")
def purge_expired_cache():
return {"deleted": cache.purge_expired()}
@router.post("/cache/purge-all")
def purge_all_cache():
return {"deleted": cache.purge_all()}
@router.post("/search")
def search(request: SearchRequest):
return _handle(
lambda: search_service.search(
query=request.query,
context=request.context,
system_prompt=request.system_prompt,
max_attempts=request.max_attempts,
timeout_seconds=request.timeout_seconds,
temperature=request.temperature,
max_tokens=request.max_tokens,
use_cache=request.use_cache,
)
)
@router.get("/search")
def search_get(
q: str = Query(..., min_length=1, max_length=2000),
max_attempts: int | None = Query(None, ge=1, le=10),
timeout_seconds: int | None = Query(None, ge=30, le=600),
temperature: float | None = Query(None, ge=0, le=2),
max_tokens: int | None = Query(None, ge=256, le=16000),
use_cache: bool = Query(True),
):
return _handle(
lambda: _search_get_response(q, max_attempts, timeout_seconds, temperature, max_tokens, use_cache)
)
@router.get("/ai-search-hub/sites")
def ai_search_hub_sites():
return ai_search_hub_service.list_sites()
@router.post("/ai-search-hub/search")
def ai_search_hub_search(request: AISearchHubRequest):
return _handle(
lambda: ai_search_hub_service.search(
query=request.query,
sites=request.sites,
timeout=request.timeout,
headless=request.headless,
use_cache=request.use_cache,
)
)
@router.get("/ai-search-hub/search")
def ai_search_hub_search_get(
q: str = Query(..., min_length=1, max_length=2000),
sites: str | None = Query(None, description="Comma-separated site keys"),
timeout: int | None = Query(None, ge=30, le=300),
headless: bool | None = Query(None),
use_cache: bool = Query(True),
):
site_list = [s.strip() for s in sites.split(",") if s.strip()] if sites else None
return _handle(
lambda: ai_search_hub_service.search(
query=q,
sites=site_list,
timeout=timeout,
headless=headless,
use_cache=use_cache,
)
)
@router.get("/stocks/{stock_code}/quote")
def stock_quote(stock_code: str):
return _handle(lambda: market_data_service.stock_quote(stock_code))
@router.get("/stocks/{stock_code}/order-book")
def stock_order_book(stock_code: str):
return _handle(lambda: market_data_service.stock_order_book(stock_code))
@router.get("/stocks/{stock_code}/daily")
def stock_daily(
stock_code: str,
days: int = Query(60, ge=1, le=5000),
start_date: str | None = None,
end_date: str | None = None,
adjust: str = Query("qfq", pattern="^(|qfq|hfq)$"),
):
return _handle(lambda: market_data_service.stock_daily(stock_code, days, start_date, end_date, adjust))
@router.get("/stocks/{stock_code}/technical")
def stock_technical(
stock_code: str,
days: int = Query(120, ge=30, le=5000),
history_days: int = Query(0, ge=0, le=120),
adjust: str = Query("qfq", pattern="^(|qfq|hfq)$"),
):
return _handle(lambda: market_data_service.stock_technical(stock_code, days, history_days, adjust))
@router.get("/stocks/{stock_code}/chip")
def stock_chip(
stock_code: str,
adjust: str = Query("qfq", pattern="^(|qfq|hfq)$"),
):
return _handle(lambda: market_data_service.stock_chip(stock_code, adjust))
@router.get("/stocks/{stock_code}/chip-simple")
def stock_chip_simple(
stock_code: str,
lookback_days: int = Query(60, ge=10, le=500),
adjust: str = Query("qfq", pattern="^(|qfq|hfq)$"),
):
return _handle(lambda: market_data_service.stock_chip_simple(stock_code, lookback_days, adjust))
@router.get("/stocks/{stock_code}/fund-flow")
def stock_fund_flow(stock_code: str, days: int = Query(10, ge=1, le=120)):
return _handle(lambda: market_data_service.stock_fund_flow(stock_code, days))
@router.get("/stocks/{stock_code}/margin")
def stock_margin(stock_code: str, days: int = Query(30, ge=1, le=500)):
return _handle(lambda: market_data_service.stock_margin(stock_code, days))
@router.get("/stocks/{stock_code}/shareholders")
def stock_shareholders(stock_code: str, limit: int = Query(12, ge=1, le=50)):
return _handle(lambda: market_data_service.stock_shareholders(stock_code, limit))
@router.get("/stocks/{stock_code}/shareholder-top")
def stock_shareholder_top(stock_code: str, date: str | None = None):
return _handle(lambda: market_data_service.stock_shareholder_top(stock_code, date))
@router.get("/stocks/{stock_code}/f10/company")
def stock_f10_company(stock_code: str):
return _handle(lambda: market_data_service.stock_f10_company(stock_code))
@router.get("/stocks/{stock_code}/research-reports")
def stock_research_reports(stock_code: str, limit: int = Query(20, ge=1, le=100)):
return _handle(lambda: market_data_service.stock_research_reports(stock_code, limit))
@router.get("/fund-flow/rank")
def fund_flow_rank(
indicator: str = Query("5日", pattern="^(今日|3日|5日|10日)$"),
limit: int = Query(100, ge=1, le=5000),
):
return _handle(lambda: market_data_service.fund_flow_rank(indicator, limit))
@router.get("/capital/big-deal")
def big_deal(limit: int = Query(100, ge=1, le=5000)):
return _handle(lambda: market_data_service.big_deal(limit))
@router.get("/market/indices")
def market_indices(limit: int = Query(100, ge=1, le=1000)):
return _handle(lambda: market_data_service.market_indices(limit))
@router.get("/market/moves")
def market_moves(
move_type: str = Query("surge", pattern="^(surge|drop|change_up|change_down|mainflow|turnover|up|down|rise|fall|fund|capital|active)$"),
limit: int = Query(50, ge=1, le=200),
):
return _handle(lambda: market_data_service.market_moves(move_type, limit))
@router.get("/market/longhubang")
def market_longhubang(
date: str | None = None,
limit: int = Query(50, ge=1, le=200),
page: int = Query(1, ge=1, le=100),
):
return _handle(lambda: market_data_service.longhubang(date, limit, page))
@router.get("/market/limit-up")
def market_limit_up(date: str | None = None, limit: int = Query(100, ge=1, le=1000)):
return _handle(lambda: market_data_service.limit_pool("up", date, limit))
@router.get("/market/limit-down")
def market_limit_down(date: str | None = None, limit: int = Query(100, ge=1, le=1000)):
return _handle(lambda: market_data_service.limit_pool("down", date, limit))
@router.get("/market/breadth")
def market_breadth():
return _handle(lambda: market_data_service.market_breadth())
@router.get("/market/temperature")
def market_temperature(date: str | None = None):
return _handle(lambda: market_data_service.market_temperature(date))
@router.get("/market/margin")
def market_margin(date: str | None = None, limit: int = Query(50, ge=1, le=5000)):
return _handle(lambda: market_data_service.market_margin(date, limit))
@router.get("/market/northbound")
def market_northbound(days: int = Query(30, ge=1, le=500)):
return _handle(lambda: market_data_service.northbound_hist(days))
@router.get("/market/northbound/realtime")
def market_northbound_realtime():
return _handle(lambda: market_data_service.northbound_realtime())
@router.get("/market/northbound/holdings")
def market_northbound_holdings(
stock_code: str = Query("", description="个股代码,为空时返回市场汇总"),
limit: int = Query(50, ge=1, le=500),
):
return _handle(lambda: market_data_service.northbound_holdings(stock_code, limit))
@router.get("/market/shenwan-industry")
def shenwan_industry(limit: int = Query(50, ge=1, le=200)):
return _handle(lambda: market_data_service.shenwan_industry(limit))
@router.get("/market/fund-holdings")
def fund_holdings(date: str = Query("20260331", description="报告期,格式 YYYYMMDD"), limit: int = Query(100, ge=1, le=5000)):
return _handle(lambda: market_data_service.fund_holdings(date, limit))
@router.get("/market/fund-structure")
def fund_structure():
return _handle(lambda: market_data_service.fund_structure())
@router.get("/boards/flow")
def board_flow(
category: str = Query("industry", pattern="^(industry|concept|region|concepts|area|province)$"),
limit: int = Query(100, ge=1, le=200),
):
return _handle(lambda: market_data_service.board_flow(category, limit))
@router.get("/boards/concepts/flow")
def concept_flow(
symbol: str = Query("即时"),
limit: int = Query(1000, ge=1, le=5000),
):
return _handle(lambda: market_data_service.concept_flow(symbol, limit))
@router.get("/boards/industries/flow")
def industry_flow(
symbol: str = Query("即时"),
limit: int = Query(1000, ge=1, le=5000),
):
return _handle(lambda: market_data_service.industry_flow(symbol, limit))
@router.get("/indices/{index_code}/fund-flow")
def index_fund_flow(
index_code: str,
interval: str = Query("1m", pattern="^(1|1m|5|5m|15|15m|30|30m|60|60m|101|1d|day|daily)$"),
limit: int = Query(120, ge=1, le=2000),
):
return _handle(lambda: market_data_service.index_fund_flow(index_code, interval, limit))
@router.get("/etfs/spot")
def etf_spot(limit: int = Query(200, ge=1, le=5000)):
return _handle(lambda: market_data_service.etf_spot(limit))
@router.get("/etfs/premium")
def etf_premium(
limit: int = Query(200, ge=1, le=5000),
sort: str = Query("abs", pattern="^(abs|premium|discount|code)$"),
):
return _handle(lambda: market_data_service.etf_premium(limit, sort))
@router.get("/etfs/{fund_code}/quote")
def etf_quote(fund_code: str):
return _handle(lambda: market_data_service.etf_quote(fund_code))
@router.get("/etfs/{fund_code}/premium")
def etf_premium_detail(fund_code: str):
return _handle(lambda: market_data_service.etf_premium_detail(fund_code))
@router.get("/etfs/{fund_code}/daily")
def etf_daily(
fund_code: str,
days: int = Query(120, ge=1, le=5000),
start_date: str | None = None,
end_date: str | None = None,
adjust: str = Query("", pattern="^(|qfq|hfq)$"),
):
return _handle(lambda: market_data_service.etf_daily(fund_code, days, start_date, end_date, adjust))
@router.get("/funds/open/spot")
def fund_open_spot(limit: int = Query(200, ge=1, le=5000)):
return _handle(lambda: market_data_service.fund_open_spot(limit))
@router.get("/funds/money/spot")
def fund_money_spot(limit: int = Query(200, ge=1, le=5000)):
return _handle(lambda: market_data_service.fund_money_spot(limit))
@router.get("/funds/{fund_code}/nav")
def fund_nav(
fund_code: str,
fund_type: str = Query("open", pattern="^(open|money|etf)$"),
limit: int = Query(300, ge=1, le=5000),
):
return _handle(lambda: market_data_service.fund_nav(fund_code, fund_type, limit))
@router.get("/hk/stocks/{stock_code}/short-selling")
def hk_short_selling(
stock_code: str,
limit: int = Query(100, ge=1, le=1000),
pages: int = Query(2, ge=1, le=20),
):
return _handle(lambda: market_data_service.hk_short_selling(stock_code, limit, pages))
@router.get("/legacy/boards/concept-hot")
def legacy_board_concept_hot(
lookback_days: int = Query(1, ge=1, le=60),
limit: int = Query(30, ge=1, le=100),
):
return _handle(lambda: market_data_service.legacy_board_concept_hot(lookback_days, limit))
@router.get("/legacy/boards/sector-flow")
def legacy_board_sector_flow(limit: int = Query(30, ge=1, le=100)):
return _handle(lambda: market_data_service.legacy_board_sector_flow(limit))
@router.get("/legacy/boards/temperature")
def legacy_board_temperature(date: str | None = None):
return _handle(lambda: market_data_service.legacy_board_temperature(date))
@router.get("/legacy/boards/news")
def legacy_board_news(
stock_code: str = "000300",
data_type: str = Query("global_news", pattern="^(news|global_news|notice|important_news|hotspot_news)$"),
limit: int = Query(20, ge=1, le=100),
):
return _handle(lambda: market_data_service.legacy_board_news(stock_code, data_type, limit))
@router.get("/legacy/boards/sector-ambush")
def legacy_sector_ambush(
lookback_days: int = Query(60, ge=1, le=250),
limit: int = Query(10, ge=1, le=50),
):
return _handle(lambda: market_data_service.legacy_sector_ambush(lookback_days, limit))
@router.get("/legacy/boards/sector-leaders")
def legacy_sector_leaders(
sector_name: str,
lookback_days: int = Query(5, ge=1, le=60),
top_n: int = Query(3, ge=1, le=10),
):
return _handle(lambda: market_data_service.legacy_sector_leaders(sector_name, lookback_days, top_n))
@router.get("/legacy/boards/leader-frequency")
def legacy_leader_frequency(
stock_name: str,
lookback_days: int = Query(5, ge=1, le=60),
):
return _handle(lambda: market_data_service.legacy_leader_frequency(stock_name, lookback_days))
@router.get("/news/global")
def global_news(limit: int = Query(50, ge=1, le=200)):
return _handle(lambda: market_data_service.global_news(limit))
@router.get("/stocks/{stock_code}/news")
def stock_news(stock_code: str, limit: int = Query(30, ge=1, le=200)):
return _handle(lambda: market_data_service.stock_news(stock_code, limit))
@router.get("/stocks/{stock_code}/notices")
def stock_notices(stock_code: str, limit: int = Query(30, ge=1, le=200)):
return _handle(lambda: market_data_service.stock_notices(stock_code, limit))
@router.get("/stocks/{stock_code}/financial")
def stock_financial(
stock_code: str,
kind: str = Query("abstract", pattern="^(abstract|indicators|forecast|express)$"),
limit: int = Query(20, ge=1, le=200),
report_date: str | None = None,
start_year: int | None = Query(None, ge=2000, le=2100, description="起始年份"),
end_year: int | None = Query(None, ge=2000, le=2100, description="结束年份"),
):
return _handle(lambda: market_data_service.stock_financial(stock_code, kind, limit, report_date, str(start_year) if start_year is not None else None, str(end_year) if end_year is not None else None))
@router.get("/stocks/{stock_code}/income")
def stock_income(
stock_code: str,
limit: int = Query(10, ge=1, le=200),
kind: str = Query("ytd", pattern="^(ytd|quarterly|sq|single_quarter)$"),
):
return _handle(lambda: market_data_service.stock_income(stock_code, limit, kind))
@router.get("/stocks/{stock_code}/balancesheet")
def stock_balancesheet(stock_code: str, limit: int = Query(10, ge=1, le=200)):
return _handle(lambda: market_data_service.stock_balancesheet(stock_code, limit))
@router.get("/stocks/{stock_code}/cashflow")
def stock_cashflow(
stock_code: str,
limit: int = Query(10, ge=1, le=200),
kind: str = Query("ytd", pattern="^(ytd|quarterly|sq|single_quarter)$"),
):
return _handle(lambda: market_data_service.stock_cashflow(stock_code, limit, kind))
@router.get("/stocks/{stock_code}/dividends")
def stock_dividends(
stock_code: str,
limit: int = Query(20, ge=1, le=200),
kind: str = Query("main", pattern="^(main|allotment|all)$"),
):
return _handle(lambda: market_data_service.stock_dividends(stock_code, limit, kind))
@router.get("/stocks/{stock_code}/equity-history")
def stock_equity_history(stock_code: str, limit: int = Query(20, ge=1, le=200)):
return _handle(lambda: market_data_service.stock_equity_history(stock_code, limit))
@router.get("/stocks/{stock_code}/freeholders")
def stock_freeholders(
stock_code: str,
limit: int = Query(20, ge=1, le=200),
end_date: str | None = None,
):
return _handle(lambda: market_data_service.stock_freeholders(stock_code, limit, end_date))
@router.get("/stocks/{stock_code}/daily-basic")
def stock_daily_basic(stock_code: str, days: int = Query(30, ge=1, le=500)):
return _handle(lambda: market_data_service.stock_daily_basic(stock_code, days))
@router.get("/market/trade-calendar")
def trade_calendar(
start_date: str | None = None,
end_date: str | None = None,
limit: int = Query(5000, ge=1, le=10000),
):
return _handle(lambda: market_data_service.trade_calendar(start_date, end_date, limit))
@router.get("/macro/chinabond/yield-curve")
def china_bond_yield_curve(
start_date: str | None = None,
end_date: str | None = None,
days: int = Query(30, ge=1, le=3650),
limit: int = Query(500, ge=1, le=5000),
):
return _handle(lambda: market_data_service.china_bond_yield_curve(start_date, end_date, days, limit))
@router.get("/macro/china/{indicator}")
def china_macro(indicator: str, limit: int = Query(200, ge=1, le=5000)):
return _handle(lambda: market_data_service.china_macro(indicator, limit))
# ---- US market (sourced from niuone patterns) ----
@router.get("/us/indices")
def us_indices(limit: int = Query(20, ge=1, le=100)):
return _handle(lambda: market_data_service.us_indices(limit))
@router.get("/us/sectors")
def us_sectors(limit: int = Query(20, ge=1, le=50)):
return _handle(lambda: market_data_service.us_sectors(limit))
@router.get("/us/market-summary")
def us_market_summary():
return _handle(lambda: market_data_service.us_market_summary())
@router.get("/us/stocks/{symbol}/quote")
def us_stock_quote(symbol: str):
return _handle(lambda: market_data_service.us_stock_quote(symbol))
@router.get("/us/stocks/{symbol}/daily")
def us_stock_daily(symbol: str, days: int = Query(60, ge=1, le=2000)):
return _handle(lambda: market_data_service.us_stock_daily(symbol, days))
# ---- X / Twitter timeline (sourced from niuone patterns) ----
@router.get("/social/x/timeline")
def x_timeline(
accounts: str = Query(..., min_length=1, max_length=2000, description="X 账号,逗号分隔,如 elonmusk,OpenAI"),
limit: int = Query(5, ge=1, le=10, description="每个账号最多条数"),
hydrate: bool = Query(True, description="是否补充推文上下文/媒体直链"),
):
return _handle(lambda: market_data_service.x_timeline(accounts, limit, hydrate))
@router.get("/market/futures-basis")
def futures_basis(days: int = Query(1, ge=1, le=120)):
return _handle(lambda: market_data_service.futures_basis(days))
@router.get("/admin/source-pool-stats")
def source_pool_stats():
"""获取源调用池的统计信息(用于监控 worker pool 使用情况)"""
stats = get_source_pool_stats()
return {
"pool_stats": stats,
"recommendations": {
"worker_pool_saturated": stats.get("active", 0) >= stats.get("max_concurrent", 2),
"suggestion": "如果 active >= max_concurrent,说明 worker pool 已饱和,需要增加 MAX_CONCURRENT_SOURCES 或优化上游响应时间",
},
}
@router.get("/admin/health-detailed")
def health_detailed():
"""详细的健康检查,包括源调用池状态"""
pool_stats = get_source_pool_stats()
return {
"ok": True,
"version": "v24-concurrent-control",
"source_pool": pool_stats,
"config": {
"max_concurrent_sources": int(os.getenv("MAX_CONCURRENT_SOURCES", "2")),
"source_pool_workers": int(os.getenv("SOURCE_POOL_WORKERS", "4")),
"source_timeout_seconds": int(os.getenv("SOURCE_TIMEOUT_SECONDS", "15")),
},
}
import os