app/api/routes.py

"""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.config import settings
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("/boards/catalog")
def boards_catalog():
    return _handle(lambda: market_data_service.boards_catalog())


@router.get("/boards/{board_code}/daily")
def board_daily(

    board_code: str,

    days: int = Query(120, ge=1, le=5000),

    start_date: str | None = None,

    end_date: str | None = None,

):
    return _handle(lambda: market_data_service.board_daily(board_code, days, start_date, end_date))


@router.get("/indices/{index_code}/daily")
def index_daily(

    index_code: str,

    days: int = Query(120, ge=1, le=5000),

    start_date: str | None = None,

    end_date: str | None = None,

):
    return _handle(lambda: market_data_service.index_daily(index_code, days, start_date, end_date))


@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("/news/cls-red-telegraph")
def cls_red_telegraph(limit: int = Query(50, ge=1, le=200)):
    """财联社红色电报直采端点（category=red 重要级别快讯流）。



    /news/global 把 cls 源排在 5 个 akshare 源之后兜底，正常取不到；

    cls_ai_report 等下游用本端点独立获取红色电报。

    """
    return _handle(lambda: market_data_service.cls_red_telegraph(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")),
        },
    }


@router.get("/admin/db-info")
def db_info():
    """返回服务运行时实际连接的数据库信息（密码脱敏）+ 行级表行数。



    用于线上诊断"服务到底连的是哪个库"以及 import 灌库进度核对。

    """
    from sqlalchemy import text, inspect
    import re as _re

    raw = settings.database_url
    # 脱敏密码：scheme://user:password@host... -> scheme://user:***@host...
    masked = _re.sub(r"(://[^:]+):[^@]+@", r"\1:***@", raw)

    info: dict = {
        "database_url_masked": masked,
        "cache_enabled": settings.cache_enabled,
        "row_level_cache_enabled": settings.row_level_cache_enabled,
        "board_index_table": settings.board_index_table,
        "index_daily_table": settings.index_daily_table,
    }

    engine = cache.engine
    if engine is None:
        info["engine"] = None
        return info

    try:
        with engine.connect() as conn:
            dialect = engine.dialect.name
            if dialect == "sqlite":
                info["server_version"] = conn.execute(text("SELECT sqlite_version()")).scalar()
                info["current_database"] = raw.split("sqlite:///")[-1] if "sqlite:///" in raw else None
            else:
                info["server_version"] = conn.execute(text("SELECT VERSION()")).scalar()
                info["current_database"] = conn.execute(text("SELECT DATABASE()")).scalar()
    except Exception as exc:  # noqa: BLE001
        info["connect_error"] = str(exc)[:200]
        return info

    try:
        names = inspect(engine).get_table_names()
        info["tables"] = names
        for t in (settings.board_index_table, settings.index_daily_table, "api_cache"):
            if t in names:
                try:
                    n = engine.connect().execute(text(f"SELECT COUNT(*) FROM {t}")).scalar()
                    info.setdefault("row_counts", {})[t] = n
                except Exception as exc:  # noqa: BLE001
                    info.setdefault("row_counts", {})[t] = f"error: {exc}"
    except Exception as exc:  # noqa: BLE001
        info["inspect_error"] = str(exc)[:200]

    return info


import os