"""市场看板数据聚合路由（/api/market-dashboard） 聚合同花顺 SDK 多个接口，一次性返回前端看板所需的全部数据： - 主要指数行情 + 估值分位（PE/PB） - 集合竞价信号（竞价基准线 + 情绪判断） - 全市场涨跌统计（上涨/下跌/平盘/涨停/跌停/炸板/成交额） - 市场温度评分（结合竞价基准校准） - 涨跌家数分布 - 行业强度榜（同花顺行业指数 + 概念板块热度） - 事件情报（连板天梯/热门股/涨停封单/炸板/飙升/龙虎榜） """ import asyncio import math import time from datetime import datetime, timezone, timedelta from fastapi import APIRouter, HTTPException from fastapi.responses import JSONResponse from services import fuyao_client, market_extra router = APIRouter() _NO_CACHE_HEADERS = {"Cache-Control": "no-store, no-cache, must-revalidate, max-age=0"} _cache: dict = {"data": None, "at": 0.0} _CACHE_TTL = 30 # 主要指数 thscode 列表 _MAIN_INDEX_THSCODES = [ "000001.SH", # 上证指数 "399001.SZ", # 深证成指 "399006.SZ", # 创业板指 "000688.SH", # 科创50 "000300.SH", # 沪深300 ] _INDEX_NAMES = { "000001.SH": "上证指数", "399001.SZ": "深证成指", "399006.SZ": "创业板指", "000688.SH": "科创50", "000300.SH": "沪深300", } BJT = timezone(timedelta(hours=8)) def _safe_float(val, default=0.0): try: v = float(val) return v if not math.isnan(v) and not math.isinf(v) else default except (TypeError, ValueError): return default def _calc_temperature(up: int, down: int, flat: int, median_pct: float, strong: int, weak: int, limit_up: int, limit_down: int, break_count: int = 0, auction_signal: str = "") -> dict: """市场温度评分（0-100）：综合涨跌比、中位数涨跌、强弱比、涨停活跃度、炸板率、竞价信号""" total = up + down + flat if total == 0: return {"score": 50, "label": "中性", "factors": {}} # 涨跌比得分（0-25分） advance_ratio = up / total advance_score = min(advance_ratio * 50, 25) # 中位数涨跌得分（0-20分）：-3%~+3% 映射到 0~20 median_score = max(0, min(20, (median_pct + 3) / 6 * 20)) # 强弱比得分（0-20分） sw_total = strong + weak if sw_total > 0: strong_ratio = strong / sw_total strong_score = min(strong_ratio * 40, 20) else: strong_score = 10 # 涨停活跃度得分（0-15分） limit_score = min(limit_up / 80 * 15, 15) # 炸板惩罚（0-10分）：炸板率越高越扣分 total_attempted = limit_up + break_count break_penalty = 0 if total_attempted > 0: break_rate = break_count / total_attempted break_penalty = break_rate * 10 # 炸板率 50% → 扣 5 分 # 竞价信号加成（±5分） auction_bonus = 0 if auction_signal == "强势高开": auction_bonus = 5 elif auction_signal == "偏强": auction_bonus = 2 elif auction_signal == "弱势低开": auction_bonus = -5 elif auction_signal == "偏弱": auction_bonus = -2 raw = advance_score + median_score + strong_score + limit_score - break_penalty + auction_bonus score = round(max(0, min(100, raw)), 1) if score >= 80: label = "强势" elif score >= 60: label = "偏强" elif score >= 40: label = "中性" elif score >= 20: label = "偏弱" else: label = "弱势" factors = { "advanceScore": round(advance_score, 1), "medianScore": round(median_score, 1), "strongScore": round(strong_score, 1), "limitScore": round(limit_score, 1), "breakPenalty": round(-break_penalty, 1), "auctionBonus": auction_bonus, } return {"score": score, "label": label, "factors": factors} async def _build_dashboard() -> dict: now = time.time() if _cache["data"] is not None and now - _cache["at"] < _CACHE_TTL: return _cache["data"] # ── 第一批并行拉取（核心数据） ── try: ( index_data, all_stocks, limit_up_data, limit_down_data, limit_break_data, industry_catalog, concept_catalog, auction_benchmark, ) = await asyncio.gather( fuyao_client.index_prices_snapshot(",".join(_MAIN_INDEX_THSCODES)), fuyao_client.prices_snapshot_all(limit=5000), fuyao_client.limit_up_pool(page=1, size=200), fuyao_client.limit_down_pool(page=1, size=1), fuyao_client.limit_break_pool(page=1, size=1), fuyao_client.index_catalog("industry"), fuyao_client.index_catalog("cn_concept"), fuyao_client.auction_short_term_benchmark(), return_exceptions=True, ) except Exception as e: raise HTTPException(status_code=502, detail=f"市场数据拉取失败: {e}") # ── 第二批并行拉取（事件数据，依赖第一批结果较轻） ── try: ( hot_stock_data, dragon_tiger_data, anomaly_data, limit_ladder_data, skyrocket_data, sector_flow_data, margin_data, global_markets_data, ) = await asyncio.gather( fuyao_client.hot_stock_list("day"), fuyao_client.dragon_tiger_list("all"), fuyao_client.anomaly_analysis_list(["SHARP_RISE", "RAPID_RALLY", "LIMIT_UP"]), fuyao_client.limit_up_ladder(), fuyao_client.skyrocket_list("day"), market_extra.fetch_sector_fund_flow(), market_extra.fetch_margin_summary(), market_extra.fetch_global_markets(), return_exceptions=True, ) except Exception: hot_stock_data = {} dragon_tiger_data = {} anomaly_data = {} limit_ladder_data = {} skyrocket_data = {} sector_flow_data = None margin_data = None global_markets_data = None # ── 解析指数 ── indices = [] if isinstance(index_data, list): for item in index_data: code = item.get("thscode", "") indices.append({ "code": code, "name": _INDEX_NAMES.get(code, item.get("name", code)), "price": _safe_float(item.get("last_price")), "change": _safe_float(item.get("price_change")), "changePct": _safe_float(item.get("price_change_ratio_pct")), "prevClose": _safe_float(item.get("prev_price")), "turnover": _safe_float(item.get("turnover")), }) # ── 指数估值（PE/PB） — 同花顺估值API仅支持个股，指数暂不支持 ── # ── 解析全市场涨跌统计 ── up_count = 0 down_count = 0 flat_count = 0 total_turnover = 0.0 all_changes = [] stocks_list = all_stocks if isinstance(all_stocks, list) else [] for s in stocks_list: chg = _safe_float(s.get("price_change_ratio_pct")) turnover = _safe_float(s.get("turnover")) total_turnover += turnover if chg > 0: up_count += 1 elif chg < 0: down_count += 1 else: flat_count += 1 all_changes.append(chg) total_stocks = up_count + down_count + flat_count median_change = sorted(all_changes)[len(all_changes) // 2] if all_changes else 0 # 涨停/跌停/炸板 精确统计 limit_up_count = 0 limit_down_count = 0 break_count = 0 if isinstance(limit_up_data, dict): pagination = limit_up_data.get("pagination", {}) limit_up_count = pagination.get("total", 0) or len(limit_up_data.get("item", [])) if isinstance(limit_down_data, dict): pagination = limit_down_data.get("pagination", {}) limit_down_count = pagination.get("total", 0) or len(limit_down_data.get("item", [])) if isinstance(limit_break_data, dict): pagination = limit_break_data.get("pagination", {}) break_count = pagination.get("total", 0) or len(limit_break_data.get("item", [])) # 强势/弱势（涨幅 > 2% 为强，< -2% 为弱） strong_count = sum(1 for c in all_changes if c > 2) weak_count = sum(1 for c in all_changes if c < -2) # 市场宽度 market_breadth = round(up_count / total_stocks * 100, 1) if total_stocks > 0 else 50 # ── 集合竞价信号 ── auction_signal = "" auction_detail = {} if isinstance(auction_benchmark, dict) and not isinstance(auction_benchmark, Exception): # 短线风向标：根据竞价基准判断多空 benchmark_score = _safe_float(auction_benchmark.get("score", 0)) benchmark_label = auction_benchmark.get("label", "") auction_detail = { "score": benchmark_score, "label": benchmark_label, "date": auction_benchmark.get("date", ""), "benchmark": auction_benchmark.get("benchmark", {}), } if benchmark_score >= 60: auction_signal = "强势高开" elif benchmark_score >= 45: auction_signal = "偏强" elif benchmark_score <= 30: auction_signal = "弱势低开" elif benchmark_score <= 45: auction_signal = "偏弱" else: auction_signal = "中性" # ── 市场温度（结合竞价信号校准） ── temperature = _calc_temperature( up_count, down_count, flat_count, median_change, strong_count, weak_count, limit_up_count, limit_down_count, break_count, auction_signal, ) market_stats = { "upCount": up_count, "downCount": down_count, "flatCount": flat_count, "total": total_stocks, "marketBreadth": market_breadth, "medianChange": round(median_change, 2), "strongCount": strong_count, "weakCount": weak_count, "limitUp": limit_up_count, "limitDown": limit_down_count, "limitBreak": break_count, "breakRate": round(break_count / (limit_up_count + break_count) * 100, 1) if (limit_up_count + break_count) > 0 else 0, "totalTurnover": round(total_turnover, 2), "temperature": temperature, "auction": auction_detail, "auctionSignal": auction_signal, } # 两融余额（交易所 T+1 披露），供看板展示、AI 分析与次日环比快照使用 if isinstance(margin_data, dict): market_stats["marginBalanceYi"] = margin_data.get("balanceYi") market_stats["marginChangeYi"] = margin_data.get("changeYi") market_stats["marginDate"] = margin_data.get("date") # ── 行业强度榜（使用行业指数真实涨幅） ── sector_strength = [] industries = industry_catalog if isinstance(industry_catalog, list) else [] industry_codes = [ind.get("thscode", "") for ind in industries[:31] if ind.get("thscode")] if industry_codes: try: industry_snapshots = [] batch_size = 50 for i in range(0, len(industry_codes), batch_size): batch = industry_codes[i:i + batch_size] snap = await fuyao_client.index_prices_snapshot(",".join(batch)) if isinstance(snap, list): industry_snapshots.extend(snap) await asyncio.sleep(0.05) for snap_item in industry_snapshots: code = snap_item.get("thscode", "") name = "" for ind in industries: if ind.get("thscode") == code: name = ind.get("name", code) break gain_pct = _safe_float(snap_item.get("price_change_ratio_pct")) last = _safe_float(snap_item.get("last_price")) sector_strength.append({ "code": code, "name": name or code, "price": last, "change": _safe_float(snap_item.get("price_change")), "changePct": gain_pct, }) sector_strength.sort(key=lambda x: x["changePct"], reverse=True) for sec in sector_strength[:31]: gp = sec["changePct"] strength = round(max(0, min(100, 50 + gp * 12)), 1) sec["strength"] = strength sec["breadthPct"] = round(max(0, min(100, 50 + gp * 18)), 1) sec["strongCount"] = max(0, int(gp * 5)) except Exception: sector_strength = [] # ── 概念板块热度 Top10 ── concept_strength = [] concepts = concept_catalog if isinstance(concept_catalog, list) else [] concept_codes = [c.get("thscode", "") for c in concepts[:20] if c.get("thscode")] if concept_codes: try: concept_snapshots = [] for i in range(0, len(concept_codes), 50): batch = concept_codes[i:i + 50] snap = await fuyao_client.index_prices_snapshot(",".join(batch)) if isinstance(snap, list): concept_snapshots.extend(snap) await asyncio.sleep(0.05) for snap_item in concept_snapshots: code = snap_item.get("thscode", "") name = "" for c in concepts: if c.get("thscode") == code: name = c.get("name", code) break concept_strength.append({ "code": code, "name": name or code, "changePct": _safe_float(snap_item.get("price_change_ratio_pct")), }) concept_strength.sort(key=lambda x: x["changePct"], reverse=True) except Exception: concept_strength = [] # ── 事件情报 ── events = [] # 1) 连板天梯 if isinstance(limit_ladder_data, dict): ladder_items = limit_ladder_data.get("item", []) if ladder_items: today_boards = ladder_items[0].get("boards", {}) for board_key in ("seven_over", "six_board", "five_board", "four_board", "three_board", "two_board"): board_list = today_boards.get(board_key, []) for item in board_list[:2]: board_num = item.get("board_num", 0) events.append({ "type": "ladder", "label": f"{board_num}连板" if board_num > 1 else "首板", "name": item.get("name", ""), "code": item.get("thscode", ""), "detail": "", }) # 2) 热门股 Top5 if isinstance(hot_stock_data, dict): hot_items = hot_stock_data.get("item", []) for item in hot_items[:5]: rank = item.get("rank", "") heat = item.get("heat", "") trend = item.get("rank_trend", "") trend_icon = "↑" if trend == "up" else ("↓" if trend == "down" else "→") events.append({ "type": "hot", "label": f"热门 #{rank}" if rank else "热门", "name": item.get("name", ""), "code": item.get("thscode", ""), "detail": f"热度 {heat} {trend_icon}", }) # 3) 飙升榜 Top3 if isinstance(skyrocket_data, dict): sky_items = skyrocket_data.get("item", []) for item in sky_items[:3]: heat = item.get("heat", "") trend = item.get("rank_trend", "") trend_icon = "↑" if trend == "up" else ("↓" if trend == "down" else "→") events.append({ "type": "skyrocket", "label": "飙升", "name": item.get("name", ""), "code": item.get("thscode", ""), "detail": f"飙升指数 {heat} {trend_icon}", }) # 4) 涨停池封单 Top3 if isinstance(limit_up_data, dict): lu_items = limit_up_data.get("item", []) for item in lu_items[:3]: seal = _safe_float(item.get("seal_money", 0)) reason = item.get("limit_up_reason", "") continue_text = item.get("continue_day_text", "") events.append({ "type": "limit_up", "label": continue_text or "涨停", "name": item.get("name", ""), "code": item.get("thscode", ""), "detail": f"封单 {seal / 10000:.0f}万　{reason}" if reason else f"封单 {seal / 10000:.0f}万", }) # 5) 炸板池 Top3 if isinstance(limit_break_data, dict): break_items = limit_break_data.get("item", []) for item in break_items[:3]: events.append({ "type": "limit_break", "label": "炸板", "name": item.get("name", ""), "code": item.get("thscode", ""), "detail": f"开板 {_safe_float(item.get('open_times', 0))}次", }) # 6) 龙虎榜 if isinstance(dragon_tiger_data, dict): dt_items = dragon_tiger_data.get("item", []) for item in dt_items[:3]: reason = item.get("reason", item.get("上榜原因", "")) events.append({ "type": "dragon_tiger", "label": "龙虎榜", "name": item.get("name", ""), "code": item.get("thscode", ""), "detail": reason[:40] if reason else "", }) # 7) 异动分析 if isinstance(anomaly_data, dict): anomaly_items = anomaly_data.get("item", []) for item in anomaly_items[:3]: tag = item.get("tagName", item.get("tag_code", "")) events.append({ "type": "anomaly", "label": "异动", "name": item.get("name", ""), "code": item.get("thscode", ""), "detail": tag, }) # 8) 盘面统计摘要（固定在最后） events.append({ "type": "summary", "label": "盘面", "name": f"上涨 {up_count} / 下跌 {down_count} / 平盘 {flat_count}", "code": "", "detail": f"涨停 {limit_up_count} 跌停 {limit_down_count} 炸板 {break_count}", }) # ── 连板梯队（完整版，供 AI 分析用；上方 events 天梯只取前2保持看板精简） ── # 涨停池里带 limit_up_reason / seal_money，按代码匹配给梯队个股 reason_by_code = {} if isinstance(limit_up_data, dict): for lu in limit_up_data.get("item", []) or []: code = lu.get("thscode", "") if code: reason_by_code[code] = { "reason": (lu.get("limit_up_reason") or "").strip(), "sealWan": round(_safe_float(lu.get("seal_money")) / 10000), } _LADDER_LEVELS = [ ("seven_over", 7, "7连板+"), ("six_board", 6, "6连板"), ("five_board", 5, "5连板"), ("four_board", 4, "4连板"), ("three_board", 3, "3连板"), ("two_board", 2, "2连板"), ("first_board", 1, "首板"), ] limit_ladder = [] if isinstance(limit_ladder_data, dict): ladder_items = limit_ladder_data.get("item", []) if ladder_items: today_boards = ladder_items[0].get("boards", {}) or {} seen_keys = set() for key, board_num, label in _LADDER_LEVELS: seen_keys.add(key) # 首板数量多（几十家）只取前8家；2连板以上全量保留 cap = 8 if board_num == 1 else None entries = today_boards.get(key, []) or [] if cap is not None: entries = entries[:cap] for item in entries: code = item.get("thscode", "") extra = reason_by_code.get(code, {}) limit_ladder.append({ "board": board_num, "label": label, "name": item.get("name", ""), "code": code, "reason": extra.get("reason", ""), "sealWan": extra.get("sealWan"), }) # 兜底：天梯返回了未知层级 key 时也带上（跳过已处理的已知 key） for key, entries in today_boards.items(): if key in seen_keys: continue for item in entries or []: code = item.get("thscode", "") extra = reason_by_code.get(code, {}) limit_ladder.append({ "board": _safe_float(item.get("board_num")) or 1, "label": f"{int(_safe_float(item.get('board_num')) or 1)}连板", "name": item.get("name", ""), "code": code, "reason": extra.get("reason", ""), "sealWan": extra.get("sealWan"), }) # ── 组装结果 ── result = { "indices": indices, "marketStats": market_stats, "sectorStrength": sector_strength[:31], "conceptStrength": concept_strength[:10], "sectorFundFlow": sector_flow_data if isinstance(sector_flow_data, dict) else None, "globalMarkets": global_markets_data if isinstance(global_markets_data, dict) else None, "events": events, "limitLadder": limit_ladder, "updateTime": datetime.now(BJT).strftime("%Y-%m-%d %H:%M:%S"), } _cache["data"] = result _cache["at"] = now return result @router.get("/market-dashboard", summary="市场看板数据聚合") async def get_market_dashboard(): try: data = await _build_dashboard() return JSONResponse({"data": data}, headers=_NO_CACHE_HEADERS) except HTTPException: raise except Exception as e: raise HTTPException(status_code=500, detail=f"市场看板数据获取失败: {e}")