astock.py

# -*- coding: utf-8 -*-
"""
astock.py — 基于 "A股全栈数据(a-stock-data)" skill 的直连配方实现的深度数据层。

设计原则（与该 skill 一致，并贴合本项目"少用东方财富、多用腾讯/同花顺"的诉求）：
  * 行情/估值/财务 优先走 腾讯财经 / 同花顺 / 新浪 —— 这些不封 IP、无需 key。
  * 东方财富(eastmoney) 仅用于其「独有、别处拿不到」的数据（龙虎榜/融资融券/解禁/资金流/
    研报/公告），且全部经 em_get() 串行限流，避免被风控中断。
  * 所有函数都带超时 + 优雅降级：失败返回 {"source":"fail"/"demo", "note":...} 而不是抛异常，
    保证前端任何一个面板挂了都不影响全局灯。

被 server.py 调用，不依赖本文件以外任何项目内状态。
"""
import os, re, json, time, random
from datetime import datetime, timedelta

UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/117.0.0.0 Safari/537.36"

# ── requests 兼容（server 环境已装；这里做兜底） ──────────────────────────
try:
    import requests
    _HAVE_REQ = True
except Exception:
    import urllib.request as _ul
    _HAVE_REQ = False

def _get(url, params=None, headers=None, timeout=12, **kw):
    """统一 GET：优先 requests，缺失则 urllib 兜底。返回类 requests.Response 对象(有 .text/.json()/.status_code)。"""
    hdrs = {"User-Agent": UA}
    if headers: hdrs.update(headers)
    if _HAVE_REQ:
        return requests.get(url, params=params, headers=hdrs, timeout=timeout, **kw)
    # urllib 兜底
    import urllib.parse as _up
    full = url + ("?" + _up.urlencode(params) if params else "")
    req = _ul.Request(full, headers=hdrs)
    try:
        with _ul.urlopen(req, timeout=timeout) as r:
            class _R:
                pass
            o = _R(); o.status_code = r.status; o.text = r.read().decode("utf-8", "ignore")
            def _j(): return json.loads(o.text)
            o.json = _j; return o
    except Exception as e:
        class _R:
            status_code = 0; text = ""
            def json(self): return {}
        o = _R(); o.note = str(e); return o

# ── 东财限流封装（所有 eastmoney.com 请求必经此处） ───────────────────────
EM_SESSION = None
EM_MIN_INTERVAL = 1.2          # 两次东财请求最小间隔(秒)，批量可调大
_em_last_call = [0.0]

def _em_session():
    global EM_SESSION
    if EM_SESSION is None:
        if _HAVE_REQ:
            EM_SESSION = requests.Session()
            EM_SESSION.headers.update({"User-Agent": UA})
        else:
            EM_SESSION = None
    return EM_SESSION

def em_get(url, params=None, headers=None, timeout=15, **kw):
    """东财统一请求入口：自动节流 + 会话复用 + 默认 UA。"""
    wait = EM_MIN_INTERVAL - (time.time() - _em_last_call[0])
    if wait > 0:
        time.sleep(wait + random.uniform(0.05, 0.45))
    try:
        if _EM_SESSION_OK():
            r = EM_SESSION.get(url, params=params, headers=headers, timeout=timeout, **kw)
        else:
            r = _get(url, params=params, headers=headers, timeout=timeout, **kw)
        return r
    finally:
        _em_last_call[0] = time.time()

def _EM_SESSION_OK():
    return _HAVE_REQ and _em_session() is not None

def _now():
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

def _market_prefix(code):
    if code.startswith(("6", "9")): return "sh"
    if code.startswith("8"): return "bj"
    return "sz"

# =====================================================================
# 1) 同花顺当日强势股 + 题材归因（零鉴权，无东财依赖）
# =====================================================================
def ths_hot_reason(date=None):
    if date is None:
        date = datetime.today().strftime("%Y-%m-%d")
    url = (f"http://zx.10jqka.com.cn/event/api/getharden/"
           f"date/{date}/orderby/date/orderway/desc/charset/GBK/")
    try:
        r = _get(url, headers={"User-Agent": UA}, timeout=10)
        if r.status_code != 200:
            return {"source": "fail", "date": date, "count": 0,
                    "rows": [], "note": f"同花顺热点返回 HTTP {r.status_code}"}
        d = r.json()
        if d.get("errocode", 0) != 0:
            return {"source": "fail", "date": date, "count": 0, "rows": [],
                    "note": f"同花顺热点错误: {d.get('errormsg','')}"}
        rows = (d.get("data") or [])
        out = []
        for x in rows[:60]:
            reason = x.get("reason") or ""
            out.append({
                "code": str(x.get("code", "")),
                "name": x.get("name", ""),
                "change_pct": float(x.get("zhangfu") or 0),
                "close": float(x.get("close") or 0),
                "turnover_pct": float(x.get("huanshou") or 0),
                "reason": reason,
                "tags": [t.strip() for t in str(reason).split("+") if t.strip()],
            })
        # 题材热度聚合（前 12）
        from collections import Counter
        cnt = Counter()
        for x in out:
            cnt.update(x["tags"])
        return {"source": "live-10jqka", "date": date, "count": len(out),
                "rows": out, "topThemes": cnt.most_common(12),
                "updatedAt": _now(), "note": f"同花顺当日强势股 {len(out)} 只（含题材归因）"}
    except Exception as e:
        return {"source": "fail", "date": date, "count": 0, "rows": [],
                "note": f"同花顺热点拉取失败: {str(e)[:80]}"}

# ── 同花顺强势股概念缓存（只算一次，供所有个股板块标签使用） ─────────────
_THS_HOT_CACHE = None
_THS_HOT_TS = 0

def _ths_hot_cached():
    global _THS_HOT_CACHE, _THS_HOT_TS
    if _THS_HOT_CACHE is not None and time.time() - _THS_HOT_TS < 120:
        return _THS_HOT_CACHE
    _THS_HOT_CACHE = ths_hot_reason()
    _THS_HOT_TS = time.time()
    return _THS_HOT_CACHE

# 内置兜底映射：灰/弱相关概念（非真实数据，仅 akshare/同花顺都拿不到时）
_DEMO_SECTORS_MAP = {
    "600519": ["白酒", "食品饮料", "MSCI中国"], "000858": ["白酒", "食品饮料"], "000568": ["白酒"],
    "600036": ["银行", "跨境支付(CIPS)"], "601398": ["银行", "中特估"], "601318": ["保险"],
    "300750": ["电池", "储能", "新能源车"], "002594": ["汽车整车", "新能源车", "比亚迪概念"],
    "300059": ["证券", "互联网金融"], "600030": ["证券"], "600276": ["化学制药", "创新药"],
    "000333": ["家电", "智能家居"], "600900": ["电力", "水电", "高股息"], "601012": ["光伏设备"],
    "688981": ["半导体", "芯片", "中芯国际概念", "科创板"], "603501": ["半导体", "芯片"],
    "688041": ["半导体", "芯片", "AI芯片"], "688256": ["半导体", "芯片", "存储芯片"],
    "002049": ["半导体", "芯片", "存储芯片"], "000002": ["房地产"], "600048": ["房地产"],
    "601899": ["工业金属", "黄金", "铜"], "300760": ["医疗器械", "医疗新基建"],
    "002230": ["软件开发", "人工智能", "科大讯飞概念"], "300803": ["软件开发", "信创"],
}

def stock_sectors(code):
    """
    返回个股所属板块/概念列表，按相关性分三级：
      rank=1 紫色：主行业（akshare 东财行业 / 内置兜底首项）
      rank=2 蓝色：当日同花顺强势股题材 tags（真实热点）
      rank=3 灰色：其余兜底概念 / 弱相关
    每个元素: {"name": str, "rank": int, "url": str}
    """
    code = str(code)
    out = []          # (rank, name)
    seen = set()
    # 1) 主行业（紫色）— 优先 akshare 东财个股资料
    try:
        import akshare as ak
        info = ak.stock_individual_info_em(symbol=code)
        if isinstance(info, dict):
            ind = info.get("行业")
            if ind and str(ind) not in seen:
                out.append((1, str(ind)))
                seen.add(str(ind))
    except Exception:
        pass
    # 2) 同花顺当日强势股 tags（蓝色）— 热点归因，真实相关
    hot = _ths_hot_cached()
    if hot and hot.get("source") == "live-10jqka":
        for row in hot.get("rows") or []:
            if row.get("code") == code:
                for t in (row.get("tags") or []):
                    if t and t not in seen:
                        out.append((2, t))
                        seen.add(t)
                # 该股本身是强势股，主 reason 也作为紫色相关行业？保留为蓝色题材
                break
    # 3) 兜底映射（灰色）
    for nm in _DEMO_SECTORS_MAP.get(code, []):
        if nm not in seen:
            out.append((3, nm))
            seen.add(nm)
    # 拼装 URL：我们自己的板块页优先；同时保留外部官网链接供特定场景使用
    rows = []
    for rank, nm in out:
        import urllib.parse as _up
        own_url = "#/sector/" + _up.quote(nm)
        ths_url = "https://basic.10jqka.com.cn/" + _up.quote(nm) + "/"
        em_url = "https://so.eastmoney.com/news/s?keyword=" + _up.quote(nm)
        rows.append({
            "name": nm, "rank": rank,
            "url": own_url,          # 个股/热门题材板块标签默认跳自建板块页
            "url_ths": ths_url,      # 需要跳官网时可用
            "url_em": em_url
        })
    return {"source": "live-10jqka" if (hot and hot.get("source") == "live-10jqka") else "demo",
            "code": code, "sectors": rows,
            "note": "紫色=主行业，蓝色=当日热点题材，灰色=兜底/弱相关；点击跳转到自建板块页"}

# =====================================================================
# 2) 北向资金（同花顺 hexin.cn，无东财依赖；上游 2024-08 后部分断供）
# =====================================================================
def northbound_realtime():
    url = "https://data.hexin.cn/market/hsgtApi/method/dayChart/"
    try:
        r = _get(url, headers={"User-Agent": UA, "Host": "data.hexin.cn",
                                "Referer": "https://data.hexin.cn/"}, timeout=10)
        if r.status_code != 200:
            return {"source": "fail", "hgt": None, "sgt": None,
                    "note": f"北向接口返回 HTTP {r.status_code}（上游可能暂停）"}
        d = r.json()
        times = d.get("time", [])
        hgt = d.get("hgt", [])
        sgt = d.get("sgt", [])
        n = len(times)
        if n == 0:
            return {"source": "fail", "hgt": None, "sgt": None,
                    "note": "北向数据上游暂未提供（同花顺 hexin 返回空，属上游断供，非本机问题）"}
        hgt_v = next((v for v in reversed(hgt) if v is not None), None)
        sgt_v = next((v for v in reversed(sgt) if v is not None), None)
        return {"source": "live-hexin", "time": times[-1],
                "hgt": hgt_v, "sgt": sgt_v,
                "updatedAt": _now(),
                "note": "沪深股通当日实时累计净买入(亿元)；若上游断供则显示最近可得值"}
    except Exception as e:
        return {"source": "fail", "hgt": None, "sgt": None,
                "note": f"北向资金拉取失败: {str(e)[:80]}"}

# =====================================================================
# 3) 个股资金流向（东财 push2，分钟级 + 120日，限流）
# =====================================================================
def eastmoney_fund_flow_minute(code):
    secid = f"1.{code}" if code.startswith("6") else f"0.{code}"
    url = "https://push2.eastmoney.com/api/qt/stock/fflow/kline/get"
    params = {"secid": secid, "klt": 1, "fields1": "f1,f2,f3,f7",
              "fields2": "f51,f52,f53,f54,f55,f56,f57"}
    headers = {"User-Agent": UA, "Referer": "https://quote.eastmoney.com/",
               "Origin": "https://quote.eastmoney.com"}
    try:
        r = em_get(url, params=params, headers=headers, timeout=10)
        d = r.json()
        rows = []
        for line in (d.get("data") or {}).get("klines", []) or []:
            p = line.split(",")
            if len(p) >= 6:
                rows.append({"time": p[0], "main_net": float(p[1] or 0),
                              "small_net": float(p[2] or 0), "mid_net": float(p[3] or 0),
                              "large_net": float(p[4] or 0), "super_net": float(p[5] or 0)})
        return rows
    except Exception:
        return []

def stock_fund_flow_120d(code):
    market_code = 1 if code.startswith("6") else 0
    url = "https://push2his.eastmoney.com/api/qt/stock/fflow/daykline/get"
    params = {"secid": f"{market_code}.{code}", "fields1": "f1,f2,f3,f7",
              "fields2": "f51,f52,f53,f54,f55,f56,f57,f58,f59,f60,f61,f62,f63,f64,f65", "lmt": "120"}
    headers = {"User-Agent": UA, "Referer": "https://quote.eastmoney.com/",
               "Origin": "https://quote.eastmoney.com"}
    try:
        r = em_get(url, params=params, headers=headers, timeout=15)
        d = r.json()
        rows = []
        for line in (d.get("data") or {}).get("klines", []) or []:
            p = line.split(",")
            if len(p) >= 7:
                rows.append({"date": p[0], "main_net": float(p[1] if p[1] != "-" else 0),
                              "small_net": float(p[2] if p[2] != "-" else 0),
                              "mid_net": float(p[3] if p[3] != "-" else 0),
                              "large_net": float(p[4] if p[4] != "-" else 0),
                              "super_net": float(p[5] if p[5] != "-" else 0)})
        return rows
    except Exception:
        return []

def fund_flow(code):
    minute = eastmoney_fund_flow_minute(code)
    d120 = stock_fund_flow_120d(code)
    if not minute and not d120:
        return {"source": "fail", "code": code, "minute": [], "d120": [],
                "main_net_today": None, "main_net_20d": None,
                "note": "资金流向(东财 push2)拉取失败或该票无数据"}
    last = minute[-1] if minute else None
    total20 = sum(x["main_net"] for x in d120[-20:]) if d120 else 0
    return {"source": "live-eastmoney", "code": code, "minute": minute[-60:], "d120": d120[-20:],
            "main_net_today": (last["main_net"] if last else None),
            "main_net_20d": (total20 if d120 else None),
            "updatedAt": _now(),
            "note": "主力/超大单/大单/中单/小单 净流入(元)；东财 push2 限流拉取"}

# =====================================================================
# 4) 龙虎榜（东财 datacenter，限流）
# =====================================================================
DATACENTER_URL = "https://datacenter-web.eastmoney.com/api/data/v1/get"

def _eastmoney_datacenter(report_name, filter_str="", page_size=50, sort_columns="", sort_types="-1"):
    params = {"reportName": report_name, "columns": "ALL", "filter": filter_str,
              "pageNumber": "1", "pageSize": str(page_size), "sortColumns": sort_columns,
              "sortTypes": sort_types, "source": "WEB", "client": "WEB"}
    r = em_get(DATACENTER_URL, params=params, timeout=15)
    d = r.json()
    if d.get("result") and d["result"].get("data"):
        return d["result"]["data"]
    return []

def dragon_tiger(code, trade_date=None):
    if trade_date is None:
        trade_date = datetime.today().strftime("%Y-%m-%d")
    try:
        start = (datetime.strptime(trade_date, "%Y-%m-%d") - timedelta(days=30)).strftime("%Y-%m-%d")
        recs = _eastmoney_datacenter("RPT_DAILYBILLBOARD_DETAILSNEW",
                                     filter_str=f"(TRADE_DATE>='{start}')(TRADE_DATE<='{trade_date}')(SECURITY_CODE=\"{code}\")",
                                     page_size=50, sort_columns="TRADE_DATE", sort_types="-1")
        records = []
        for row in recs:
            records.append({"date": str(row.get("TRADE_DATE", ""))[:10],
                             "reason": row.get("EXPLANATION", ""),
                             "net_buy": round((row.get("BILLBOARD_NET_AMT") or 0) / 1e4, 1),
                             "turnover": round(float(row.get("TURNOVERRATE") or 0), 2)})
        seats = {"buy": [], "sell": []}
        institution = {"buy_amt": 0, "sell_amt": 0, "net_amt": 0}
        if records:
            latest = records[0]["date"]
            buy = _eastmoney_datacenter("RPT_BILLBOARD_DAILYDETAILSBUY",
                                        filter_str=f"(TRADE_DATE='{latest}')(SECURITY_CODE=\"{code}\")",
                                        page_size=5, sort_columns="BUY", sort_types="-1")
            for row in buy[:5]:
                seats["buy"].append({"name": row.get("OPERATEDEPT_NAME", ""),
                                      "buy_amt": round((row.get("BUY") or 0)/1e4, 1),
                                      "sell_amt": round((row.get("SELL") or 0)/1e4, 1),
                                      "net": round((row.get("NET") or 0)/1e4, 1)})
            sell = _eastmoney_datacenter("RPT_BILLBOARD_DAILYDETAILSSELL",
                                         filter_str=f"(TRADE_DATE='{latest}')(SECURITY_CODE=\"{code}\")",
                                         page_size=5, sort_columns="SELL", sort_types="-1")
            for row in sell[:5]:
                seats["sell"].append({"name": row.get("OPERATEDEPT_NAME", ""),
                                       "buy_amt": round((row.get("BUY") or 0)/1e4, 1),
                                       "sell_amt": round((row.get("SELL") or 0)/1e4, 1),
                                       "net": round((row.get("NET") or 0)/1e4, 1)})
            for detail, side in ((buy, "buy"), (sell, "sell")):
                for row in detail:
                    if str(row.get("OPERATEDEPT_CODE", "")) == "0":
                        amt = (row.get("BUY") or 0) if side == "buy" else (row.get("SELL") or 0)
                        institution["buy_amt" if side == "buy" else "sell_amt"] += amt
            institution["buy_amt"] = round(institution["buy_amt"]/1e4, 1)
            institution["sell_amt"] = round(institution["sell_amt"]/1e4, 1)
            institution["net_amt"] = round(institution["buy_amt"] - institution["sell_amt"], 1)
        return {"source": "live-eastmoney", "code": code, "records": records, "seats": seats,
                "institution": institution, "updatedAt": _now(),
                "note": "龙虎榜(东财 datacenter)：上榜记录+买卖席位TOP5+机构动向"}
    except Exception as e:
        return {"source": "fail", "code": code, "records": [], "seats": {"buy": [], "sell": []},
                "institution": {"buy_amt": 0, "sell_amt": 0, "net_amt": 0},
                "note": f"龙虎榜拉取失败: {str(e)[:80]}"}

# =====================================================================
# 5) 研报（东财 reportapi，限流）
# =====================================================================
def research_reports(code, max_pages=3):
    try:
        all_rows = []
        for page in range(1, max_pages + 1):
            params = {"industryCode": "*", "pageSize": "100", "industry": "*", "rating": "*",
                      "ratingChange": "*", "beginTime": "2000-01-01", "endTime": "2030-01-01",
                      "pageNo": str(page), "fields": "", "qType": "0", "orgCode": "", "code": code,
                      "rcode": "", "p": str(page), "pageNum": str(page), "pageNumber": str(page)}
            r = em_get("https://reportapi.eastmoney.com/report/list", params=params,
                       headers={"Referer": "https://data.eastmoney.com/"}, timeout=30)
            d = r.json()
            rows = d.get("data") or []
            if not rows:
                break
            all_rows.extend(rows)
            if page >= (d.get("TotalPage", 1) or 1):
                break
        out = []
        for x in all_rows[:20]:
            out.append({"date": str(x.get("publishDate", ""))[:10], "org": x.get("orgSName", ""),
                        "title": x.get("title", ""), "rating": x.get("emRatingName", ""),
                        "eps_cur": x.get("predictThisYearEps"), "eps_next": x.get("predictNextYearEps"),
                        "infoCode": x.get("infoCode", "")})
        if not out:
            return {"source": "fail", "code": code, "reports": [],
                    "note": "研报(东财 reportapi)无数据（可能该票研报少或非交易日）"}
        return {"source": "live-eastmoney", "code": code, "reports": out, "updatedAt": _now(),
                "note": "研报列表(东财 reportapi)：标题/机构/评级/一致预期EPS"}
    except Exception as e:
        return {"source": "fail", "code": code, "reports": [],
                "note": f"研报拉取失败: {str(e)[:80]}"}

# =====================================================================
# 6) 公告（巨潮 cninfo，限流）
# =====================================================================
def cninfo_announcements(code, page_size=20):
    try:
        if code.startswith("6"):
            org_id = f"gssh0{code}"
        elif code.startswith(("8", "4")):
            org_id = f"gsbj0{code}"
        else:
            org_id = f"gssz0{code}"
        payload = {"stock": f"{code},{org_id}", "tabName": "fulltext", "pageSize": str(page_size),
                   "pageNum": "1", "column": "", "category": "", "plate": "", "seDate": "",
                   "searchkey": "", "secid": "", "sortName": "", "sortType": "", "isHLtitle": "true"}
        headers = {"User-Agent": UA, "Content-Type": "application/x-www-form-urlencoded",
                   "Referer": "https://www.cninfo.com.cn/new/disclosure",
                   "Origin": "https://www.cninfo.com.cn"}
        r = em_get("https://www.cninfo.com.cn/new/hisAnnouncement/query", data=payload,
                   headers=headers, timeout=15)
        d = r.json()
        out = []
        for x in (d.get("announcements") or []):
            ts = x.get("announcementTime")
            dt = datetime.fromtimestamp(ts/1000).strftime("%Y-%m-%d") if isinstance(ts, (int, float)) else str(ts)[:10]
            out.append({"title": x.get("announcementTitle", ""), "type": x.get("announcementTypeName", ""),
                        "date": dt, "url": f"https://www.cninfo.com.cn/new/disclosure/detail?annoId={x.get('announcementId','')}"})
        if not out:
            return {"source": "fail", "code": code, "anns": [],
                    "note": "公告(巨潮)无数据（非交易日或接口调整）"}
        return {"source": "live-cninfo", "code": code, "anns": out, "updatedAt": _now(),
                "note": "公告(巨潮 cninfo)：全文检索"}
    except Exception as e:
        return {"source": "fail", "code": code, "anns": [],
                "note": f"公告拉取失败: {str(e)[:80]}"}

# =====================================================================
# 7) 新浪财报三表（资产负债表/利润表/现金流量表）
# =====================================================================
def _parse_num(s):
    if s is None: return 0.0
    s = str(s).replace(",", "").strip()
    m = re.search(r"-?\d+(\.\d+)?", s)
    if not m: return 0.0
    v = float(m.group(0))
    if "亿" in s: v *= 1e8
    elif "万" in s: v *= 1e4
    return v

def sina_financial_report(code, report_type="lrb", num=6):
    prefix = "sh" if code.startswith("6") else "sz"
    url = "https://quotes.sina.cn/cn/api/openapi.php/CompanyFinanceService.getFinanceReport2022"
    params = {"paperCode": f"{prefix}{code}", "source": report_type, "type": "0",
              "page": "1", "num": str(num)}
    try:
        r = _get(url, params=params, headers={"User-Agent": UA}, timeout=15)
        if r.status_code != 200:
            return {"source": "fail", "code": code, "type": report_type, "rows": [],
                    "note": f"新浪财报返回 HTTP {r.status_code}"}
        report_list = (r.json().get("result", {}).get("data", {}).get("report_list", {}) or {})
        rows = []
        for period in sorted(report_list.keys(), reverse=True)[:num]:
            obj = report_list[period]
            rec = {"报告期": f"{period[:4]}-{period[4:6]}-{period[6:8]}"}
            for it in (obj.get("data") or []):
                title = it.get("item_title", "")
                if not title or it.get("item_value") is None:
                    continue
                rec[title] = it.get("item_value")
            rows.append(rec)
        if not rows:
            return {"source": "fail", "code": code, "type": report_type, "rows": [],
                    "note": "新浪财报三表无数据"}
        return {"source": "live-sina", "code": code, "type": report_type, "rows": rows,
                "updatedAt": _now(), "note": f"新浪财报三表({report_type})"}
    except Exception as e:
        return {"source": "fail", "code": code, "type": report_type, "rows": [],
                "note": f"新浪财报拉取失败: {str(e)[:80]}"}

# =====================================================================
# 8) 估值（腾讯行情 + 同花顺一致预期EPS，无东财依赖）
# =====================================================================
def tencent_quote_ext(codes):
    """批量腾讯行情，返回 {code: {...}}，含 PE/PB/市值/涨跌停/换手等。"""
    if isinstance(codes, str): codes = [codes]
    pref = []
    for c in codes:
        pref.append(f"{_market_prefix(c)}{c}")
    url = "https://qt.gtimg.cn/q=" + ",".join(pref)
    try:
        req = None
        if _HAVE_REQ:
            r = _get(url, timeout=10)
            data = r.text
        else:
            import urllib.request as _ul
            rq = _ul.Request(url, headers={"User-Agent": UA})
            data = _ul.urlopen(rq, timeout=10).read().decode("gbk", "ignore")
        result = {}
        for line in data.strip().split(";"):
            if '"' not in line or "=" not in line:
                continue
            key = line.split("=")[0].split("_")[-1]
            vals = line.split('"')[1].split("~")
            if len(vals) < 53:
                continue
            code = key[2:]
            result[code] = {
                "name": vals[1], "price": float(vals[3] or 0), "prevClose": float(vals[4] or 0),
                "open": float(vals[5] or 0), "change_amt": float(vals[31] or 0),
                "change_pct": float(vals[32] or 0), "high": float(vals[33] or 0),
                "low": float(vals[34] or 0), "amount_wan": float(vals[37] or 0),
                "turnover_pct": float(vals[38] or 0), "pe_ttm": float(vals[39] or 0),
                "mcap_yi": float(vals[44] or 0), "float_mcap_yi": float(vals[45] or 0),
                "pb": float(vals[46] or 0), "limit_up": float(vals[47] or 0),
                "limit_down": float(vals[48] or 0), "pe_static": float(vals[52] or 0),
            }
        return result
    except Exception:
        return {}

def _ths_eps_forecast(code):
    """同花顺一致预期EPS（解析 HTML 表格，避免 pandas 依赖）。返回 (eps_cur, eps_next, analyst_count)。"""
    url = f"https://basic.10jqka.com.cn/new/{code}/worth.html"
    try:
        r = _get(url, headers={"User-Agent": UA, "Referer": "https://basic.10jqka.com.cn/"}, timeout=15)
        html = r.text
        # 抓取所有含"每股收益"的表格行
        # 简化：找数字行。同花顺 worth 页面 EPS 预测表头含 年份/预测机构数/最小值/均值/最大值
        # 用正则抓 "年份" 行附近的数字
        # 这里采用稳健做法：找所有 <td> 数字序列，命中"每股收益"段落
        blocks = re.findall(r"每股收益.*?</table>", html, re.S)
        if not blocks:
            # 退而求其次：直接抓数值对
            nums = re.findall(r">([\d]+\.[\d]+)</td>", html)
            if len(nums) >= 3:
                return float(nums[0]), float(nums[1]), int(float(nums[2]))
            return None, None, 0
        # 在第一个含每股收益的 block 内取前几行数据
        rows = re.findall(r"<tr[^>]*>(.*?)</tr>", blocks[0], re.S)
        eps_vals = []
        for row in rows:
            tds = re.findall(r"<td[^>]*>(.*?)</td>", row, re.S)
            tds = [re.sub(r"<[^>]+>", "", t).strip() for t in tds]
            for t in tds:
                m = re.match(r"^[\d]+\.[\d]+$", t)
                if m:
                    eps_vals.append(float(t))
        if len(eps_vals) >= 2:
            return eps_vals[0], eps_vals[1], 0
        return None, None, 0
    except Exception:
        return None, None, 0

def valuation_astock(code):
    q = tencent_quote_ext([code]).get(code)
    if not q or not q.get("price"):
        return {"source": "fail", "code": code, "note": "腾讯行情未返回（本机未联网或代码无效）"}
    eps_cur, eps_next, analyst = _ths_eps_forecast(code)
    pe_ttm = q.get("pe_ttm") or 0
    pb = q.get("pb") or 0
    pe_fwd = None; cagr = 0; peg = None; digest = 0
    if eps_next:
        pe_fwd = round(q["price"] / eps_next, 2)
        if eps_cur:
            cagr = eps_next / eps_cur - 1
    elif eps_cur:
        pe_fwd = round(q["price"] / eps_cur, 2)
        if eps_cur:
            cagr = 0
    if pe_fwd and cagr > 0:
        peg = round(pe_fwd / (cagr * 100), 2)
    if pe_fwd and pe_fwd > 30 and cagr > 0:
        import math
        digest = round(math.log(pe_fwd / 30) / math.log(1 + cagr), 1)
    return {"source": "live-tencent+ths", "code": code, "name": q["name"], "price": q["price"],
            "pe_ttm": pe_ttm, "pb": pb, "mcap_yi": q["mcap_yi"],
            "eps_cur": eps_cur, "eps_next": eps_next, "analyst_count": analyst,
            "pe_fwd": pe_fwd, "cagr_pct": round(cagr * 100, 1) if cagr else None,
            "peg": peg, "digest_years": digest,
            "updatedAt": _now(),
            "note": "前向PE/PEG = 腾讯实时价 × 同花顺一致预期EPS（无需东财）"}

def finance_astock(code):
    lrb = sina_financial_report(code, "lrb", 4)
    if lrb.get("source") != "live-sina" or not lrb.get("rows"):
        return {"source": "fail", "code": code, "finance": None,
                "note": lrb.get("note", "新浪财报无数据")}
    r0 = lrb["rows"][0]
    revenue = _parse_num(r0.get("营业收入") or r0.get("营业总收入") or 0)
    net = _parse_num(r0.get("净利润") or 0)
    eps = _parse_num(r0.get("每股收益") or 0)
    rev_yoy = _parse_num(r0.get("营业收入同比增长") or r0.get("营业总收入同比增长") or 0)
    net_yoy = _parse_num(r0.get("净利润同比增长") or 0)
    return {"source": "live-sina", "code": code,
            "finance": {"reportDate": r0.get("报告期", ""),
                        "revenue": revenue, "revenueYoY": rev_yoy,
                        "netProfit": net, "netProfitYoY": net_yoy,
                        "roe": _parse_num(r0.get("净资产收益率") or 0),
                        "grossMargin": _parse_num(r0.get("销售毛利率") or 0),
                        "debtRatio": _parse_num(r0.get("资产负债率") or 0),
                        "eps": eps},
            "note": "财报(新浪三表·利润表最新期)"}

# =====================================================================
# 9) 自然语言查询（NeoData / 平安证券）— 由 server 通过子进程调用脚本
# =====================================================================
def _skill_script(skill, script):
    base = os.path.expanduser(f"~/.workbuddy/skills/{skill}/scripts/{script}")
    return base if os.path.exists(base) else None

def _run_script(cmd, timeout=45):
    """以 UTF-8 容错编码运行子进程，彻底规避中文 Windows GBK 解码失败。"""
    import subprocess, sys
    try:
        # Python 3.7+ 支持 encoding/errors；显式指定避免系统默认 GBK 解码非 ASCII 字节崩溃
        r = subprocess.run(cmd, capture_output=True, text=True,
                           encoding="utf-8", errors="replace", timeout=timeout)
        out = (r.stdout or "") + (r.stderr or "")
        return True, out
    except subprocess.TimeoutExpired as e:
        stdout = (e.stdout or "") if isinstance(e.stdout, str) else (e.stdout.decode("utf-8", "replace") if e.stdout else "")
        stderr = (e.stderr or "") if isinstance(e.stderr, str) else (e.stderr.decode("utf-8", "replace") if e.stderr else "")
        return False, f"调用超时: {stdout}{stderr}"[:200]
    except Exception as e:
        return False, f"调用失败: {str(e)[:120]}"

def neodata_query(query):
    """调用 neodata-financial-search 的 query.py。返回 (ok, text)。"""
    script = _skill_script("neodata-financial-search", "query.py")
    if not script:
        return False, "未找到 neodata-financial-search skill 的 query.py（请确认已安装该技能）"
    import sys
    return _run_script([sys.executable, script, "--query", query], timeout=45)

def pingan_news_search(question, top_k=5):
    """调用 news-search 的 get_data.py。返回 (ok, text)。"""
    script = _skill_script("news-search", "get_data.py")
    if not script:
        return False, "未找到 news-search skill 的 get_data.py"
    import sys
    return _run_script([sys.executable, script, "--question", question, "--top_k", str(top_k)], timeout=45)

if __name__ == "__main__":
    # 本地快速自测（无网络时各函数应优雅降级）
    print("ths_hot_reason:", ths_hot_reason()["source"])
    print("northbound:", northbound_realtime()["source"])
    print("valuation(000858):", valuation_astock("000858")["source"])