"""股票数据路由：搜索、行情、K线、资金流向""" from fastapi import APIRouter, Query, HTTPException import re import math from services import tencent, sina, eastmoney from models import StockSearchResult, StockQuote, KLineData, FundFlowData, FundFlowSummary router = APIRouter() @router.get("/search", summary="股票搜索") async def search_stock(keyword: str = Query(..., description="关键词：名称/代码/拼音")): keyword = keyword.strip() if not keyword: return {"data": [], "count": 0} results = await tencent.search_stock(keyword) # 降级：搜索无结果但 keyword 是 6 位数字代码，直接查行情接口 if not results and re.match(r"^\d{6}$", keyword): direct = await tencent.lookup_by_quote(keyword) if direct: results = [direct] # 去重 seen = set() unique = [] for r in results: if r["code"] not in seen: seen.add(r["code"]) unique.append(r) limited = unique[:20] return {"data": limited, "count": len(limited)} @router.get("/quote", summary="实时行情") async def stock_quote(code: str = Query(..., description="6位股票代码")): if not re.match(r"^\d{6}$", code): raise HTTPException(status_code=400, detail="股票代码格式错误，需为6位数字") data = await tencent.fetch_quote(code) if not data: raise HTTPException(status_code=404, detail="股票不存在或已停牌") return {"data": data} @router.get("/history", summary="历史K线") async def stock_history( code: str = Query(..., description="6位股票代码"), days: int = Query(90, description="天数"), ): if not re.match(r"^\d{6}$", code): raise HTTPException(status_code=400, detail="股票代码格式错误，需为6位数字") # 主数据源：腾讯 klines = await tencent.fetch_history(code, days) if len(klines) >= 2: return {"data": klines, "count": len(klines), "source": "tencent"} # 降级：新浪 sina_klines = await sina.fetch_history(code, days) if sina_klines: return {"data": sina_klines, "count": len(sina_klines), "source": "sina"} if klines: return {"data": klines, "count": len(klines), "source": "tencent"} raise HTTPException(status_code=404, detail="未获取到K线数据") @router.get("/fund-flow", summary="资金流向") async def stock_fund_flow( code: str = Query(..., description="6位股票代码"), name: str = Query("", description="股票名称"), days: int = Query(21, description="目标天数"), ): if not re.match(r"^\d{6}$", code): raise HTTPException(status_code=400, detail="股票代码格式错误，需为6位数字") # 获取股票名称（如果未提供） stock_name = name if not stock_name: quote = await tencent.fetch_quote(code) if quote: stock_name = quote.get("name", "") # MX 请求 90 天以获取尽可能多的数据，最终只取最近 days 条 fetch_days = max(90, days) # 数据源1：MX API mx_data = None if stock_name: mx_data = await eastmoney.fetch_mx_api(stock_name, fetch_days) # 数据源2：push2his 补充（MX 数据不足时尝试） push2his_data = None need_more = not mx_data or len(mx_data) < days if need_more: push2his_data = await eastmoney.fetch_push2his(code, fetch_days) # 数据源3：腾讯K线（合并收盘价、涨跌幅） kline_map = await tencent.fetch_kline_map(code, fetch_days) # ---- 解析 ---- def _build_entry(date: str, main_net: float, turnover: float, force_net: float, super_large: float, large: float, retail: float, pcts: dict) -> dict: ki = kline_map.get(date, {}) return { "date": date, "mainNetInflow": main_net, "superLargeInflow": max(0, super_large), "superLargeOutflow": abs(min(0, super_large)), "largeInflow": max(0, large), "largeOutflow": abs(min(0, large)), "mediumInflow": 0, "mediumOutflow": 0, "smallInflow": 0, "smallOutflow": 0, "mainNetInflowPercent": pcts.get("main", 0), "superLargeInflowPercent": pcts.get("superLarge", 0), "superLargeOutflowPercent": pcts.get("superLargeOut", 0), "largeInflowPercent": pcts.get("large", 0), "largeOutflowPercent": pcts.get("largeOut", 0), "mediumInflowPercent": pcts.get("medium", 0), "mediumOutflowPercent": pcts.get("mediumOut", 0), "smallInflowPercent": pcts.get("small", 0), "smallOutflowPercent": pcts.get("smallOut", 0), "turnover": turnover, "mainForceNet": force_net, "retailNet": retail, "changePercent": ki.get("changePercent", 0), "closePrice": ki.get("close", 0), } parsed_data = [] # 先解析 MX 数据（主力净额准确） if mx_data: mx_data.sort(key=lambda x: x["date"]) for item in mx_data: main_net = item["mainNetInflow"] main_force = main_net super_large = main_net * 0.4 large = main_net * 0.6 parsed_data.append(_build_entry( item["date"], main_net, item["amount"], main_force, super_large, large, 0, {}, )) # 补全 push2his 数据（f52=主力净流入为权威值） if push2his_data: existing_dates = {e["date"] for e in parsed_data} for line in push2his_data: fields = line.split(",") if len(fields) < 11: continue date = fields[0] if date in existing_dates: continue ki = kline_map.get(date, {}) def f(i): return float(fields[i]) if fields[i] else 0 main_force = f(1) # f52 主力净流入（权威值，与 MX 口径一致） super_large_net = main_force * 0.4 large_net = main_force * 0.6 retail = f(4) + f(5) # 中单+小单净流入 pcts = { "main": f(6), "superLarge": f(7), "superLargeOut": 0, "large": f(8), "largeOut": 0, "medium": f(9), "mediumOut": 0, "small": f(10), "smallOut": 0, } parsed_data.append(_build_entry( date, main_force, ki.get("turnover", 0), main_force, super_large_net, large_net, retail, pcts, )) if not parsed_data: raise HTTPException(status_code=404, detail="未获取到资金流向数据") # 统一排序，只保留最新 days 条 parsed_data.sort(key=lambda x: x["date"]) if len(parsed_data) > days: parsed_data = parsed_data[-days:] # 重新计算累计值 cumulative_main = 0 cumulative_retail = 0 for d in parsed_data: cumulative_main += d["mainForceNet"] cumulative_retail += d["retailNet"] d["cumulativeMainNet"] = cumulative_main d["cumulativeRetailNet"] = cumulative_retail total_main = sum(d["mainForceNet"] for d in parsed_data) total_turnover = sum(d["turnover"] for d in parsed_data) total_large_inflow = sum(d["largeInflow"] for d in parsed_data) positive = sum(1 for d in parsed_data if d["mainForceNet"] > 0) negative = sum(1 for d in parsed_data if d["mainForceNet"] < 0) return { "data": parsed_data, "count": len(parsed_data), "stockCode": code, "summary": { "totalMainNet": total_main, "totalTurnover": total_turnover, "totalLargeInflow": total_large_inflow, "avgDailyMainNet": total_main / len(parsed_data) if parsed_data else 0, "positiveDays": positive, "negativeDays": negative, }, }