Files
auv/backend/routes/market_dashboard.py
T

591 lines
22 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""市场看板数据聚合路由(/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}")