"""市场补充数据源（供 AI 分析与看板扩展） - fetch_news: 新浪财经 7x24 快讯 - fetch_sector_fund_flow: 东方财富板块主力资金流排行（行业/概念） - fetch_margin_summary: 东方财富两融余额汇总（T+1 数据） 均为公开接口，失败时返回 []/None，不阻塞主流程。 """ import asyncio import httpx _TIMEOUT = 10.0 _UA = ("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36") async def fetch_news(limit: int = 30) -> list[dict]: """新浪财经 7x24 快讯，返回 [{time: 'MM-DD HH:MM', content}]；失败返回 []""" try: limit = max(1, min(int(limit or 30), 50)) except (TypeError, ValueError): limit = 30 try: async with httpx.AsyncClient(timeout=_TIMEOUT, headers={"User-Agent": _UA}) as client: resp = await client.get( "https://zhibo.sina.com.cn/api/zhibo/feed", params={"page": 1, "page_size": limit, "zhibo_id": 152, "tag_id": 0}, ) resp.raise_for_status() items = resp.json()["result"]["data"]["feed"]["list"] news = [] for it in items or []: text = (it.get("rich_text") or "").strip() if not text: continue news.append({ "time": (it.get("create_time") or "")[5:16], # 'MM-DD HH:MM' "content": text[:300], }) return news except Exception: return [] _FLOW_HOSTS = ( # push2 对部分客户端有 TLS 指纹拦截（peer closed），delay 镜像同接口且稳定 "https://push2delay.eastmoney.com", "https://push2.eastmoney.com", ) async def _fetch_flow_boards(client: httpx.AsyncClient, fs: str, po: int, pz: int) -> list[dict]: """拉取一类板块的主力净流入排行。po=1 降序（净流入最多），po=0 升序（净流出最多） 查询串保持字面量 + 号（与东财网页请求一致），逐 host 尝试。 """ qs = (f"/api/qt/clist/get?fid=f62&po={po}&pz={pz}&pn=1&np=1" f"&fltt=2&invt=2&fs={fs}&fields=f12,f14,f62,f184") last_err: Exception | None = None for host in _FLOW_HOSTS: try: resp = await client.get(host + qs) resp.raise_for_status() diff = (resp.json().get("data") or {}).get("diff") or [] break except Exception as e: last_err = e diff = [] else: raise ConnectionError(f"板块资金流全部数据源失败: {last_err}") if isinstance(diff, dict): # 兼容旧版 {index: item} 结构 diff = list(diff.values()) boards = [] for d in diff: amt = d.get("f62") if not isinstance(amt, (int, float)): continue boards.append({ "code": d.get("f12", ""), "name": d.get("f14", ""), "mainNet": round(amt / 1e8, 1), # 亿元 "mainNetPct": d.get("f184"), # 主力净占比 % }) return boards _GLOBAL_INDICES = "100.NDX,100.DJIA,100.SPX,100.HSI,100.N225,100.FTSE" _FUT_MARKETS = ("m:8", "m:113", "m:142", "m:114", "m:115") # 中金所/上期所/上期能源/大商所/郑商所 async def fetch_global_markets() -> dict | None: """海外主要指数 + 国内期货主力合约；失败返回 None""" try: async with httpx.AsyncClient(timeout=_TIMEOUT, headers={"User-Agent": _UA}) as client: overseas, futures = await asyncio.gather( _fetch_overseas_indices(client), _fetch_futures_main(client), ) if not overseas and not futures: return None return {"overseas": overseas, "futures": futures} except Exception: return None async def _fetch_overseas_indices(client: httpx.AsyncClient) -> list[dict]: resp = await client.get( "https://push2delay.eastmoney.com/api/qt/ulist.np/get", params={"secids": _GLOBAL_INDICES, "fields": "f12,f14,f2,f3", "fltt": 2, "invt": 2}, ) resp.raise_for_status() diff = (resp.json().get("data") or {}).get("diff") or [] return [ {"name": d.get("f14", ""), "price": d.get("f2"), "changePct": d.get("f3")} for d in diff ] async def _fetch_futures_main(client: httpx.AsyncClient) -> list[dict]: """国内期货主力合约（名称含"主连/主力合约"），按成交额降序取前12""" import re results: list[dict] = [] for fs in _FUT_MARKETS: try: resp = await client.get( "https://push2delay.eastmoney.com/api/qt/clist/get", params={"fid": "f6", "po": 1, "pz": 200, "pn": 1, "np": 1, "fltt": 2, "invt": 2, "fs": fs, "fields": "f12,f14,f2,f3,f6"}, ) resp.raise_for_status() diff = (resp.json().get("data") or {}).get("diff") or [] except Exception: continue for d in diff: name = d.get("f14") or "" if "主连" not in name and "主力合约" not in name: continue if "次主连" in name: # 次主力合约，排除 continue amount = d.get("f6") if not isinstance(amount, (int, float)): continue results.append({ "name": re.sub(r"(主连|主力合约)$", "", name), "price": d.get("f2"), "changePct": d.get("f3"), "amountYi": round(amount / 1e8, 1), }) results.sort(key=lambda x: -x["amountYi"]) return results[:12] async def fetch_sector_fund_flow() -> dict | None: """板块主力资金流排行：行业净流入/净流出 TOP6 + 概念净流入 TOP6；失败返回 None""" try: async with httpx.AsyncClient(timeout=_TIMEOUT, headers={"User-Agent": _UA}) as client: industry_in, industry_out, concept_in = await asyncio.gather( _fetch_flow_boards(client, "m:90+t:2", 1, 6), _fetch_flow_boards(client, "m:90+t:2", 0, 6), _fetch_flow_boards(client, "m:90+t:3", 1, 6), ) return { "industryInflow": industry_in, # 主力净流入降序 "industryOutflow": industry_out, # 升序（净流出最多在前） "conceptInflow": concept_in, "unit": "亿元", } except Exception: return None async def fetch_margin_summary() -> dict | None: """沪深北两融余额汇总（交易所 T+1 披露）；失败返回 None""" try: async with httpx.AsyncClient(timeout=_TIMEOUT, headers={"User-Agent": _UA}) as client: resp = await client.get( "https://datacenter-web.eastmoney.com/api/data/v1/get", params={ "reportName": "RPTA_RZRQ_LSHJ", "columns": "ALL", "source": "WEB", "sortColumns": "dim_date", "sortTypes": "-1", "pageSize": 2, "pageNumber": 1, }, ) resp.raise_for_status() rows = ((resp.json().get("result") or {}).get("data")) or [] if not rows: return None def _balance(row: dict) -> float: return float(row.get("RZYE") or 0) + float(row.get("RQYE") or 0) latest = rows[0] prev = rows[1] if len(rows) > 1 else None balance = _balance(latest) change = (balance - _balance(prev)) if prev else None return { "date": (latest.get("DIM_DATE") or "")[:10], "balanceYi": round(balance / 1e8), # 亿元 "changeYi": round(change / 1e8) if change is not None else None, "rzjmeYi": round(float(latest.get("RZJME") or 0) / 1e8, 1), # 融资净买入 } except Exception: return None