1#!/usr/bin/env python3
2# -*- coding: utf-8 -*-
3 
4from contextlib import redirect_stderr, redirect_stdout
5from datetime import datetime, timedelta
6from io import StringIO
7import os
8from typing import Any, Dict, Optional
9 
10 
11class AkshareAdapter:
12    def __init__(self) -> None:
13        self._ak = None
14        self._import_error = None
15        try:
16            import akshare as ak  # type: ignore
17 
18            self._ak = ak
19        except Exception as exc:
20            self._import_error = str(exc)
21 
22    def _wrap(self, fn_name: str, **payload: Any) -> Dict[str, Any]:
23        return {
24            "ok": True,
25            "source": "akshare",
26            "api": fn_name,
27            "data": payload,
28        }
29 
30    def _error(self, fn_name: str, message: str) -> Dict[str, Any]:
31        return {
32            "ok": False,
33            "source": "akshare",
34            "api": fn_name,
35            "error": message,
36        }
37 
38    def _ready_or_error(self, fn_name: str) -> Optional[Dict[str, Any]]:
39        if self._ak is None:
40            return self._error(fn_name, f"akshare import failed: {self._import_error}")
41        return None
42 
43    def _to_records(self, data: Any, top_n: int = 10) -> Any:
44        if data is None:
45            return []
46 
47        if hasattr(data, "head") and hasattr(data, "to_dict"):
48            try:
49                if top_n and top_n > 0:
50                    return data.head(top_n).to_dict(orient="records")
51                return data.to_dict(orient="records")
52            except Exception:
53                return str(data)
54 
55        return data
56 
57    def _data_len(self, data: Any) -> int:
58        try:
59            return int(len(data))
60        except Exception:
61            return 0
62 
63    def _normalize_trade_date(self, value: Optional[str]) -> str:
64        if not value or value in {"today", "今日", "今天"}:
65            return datetime.now().strftime("%Y%m%d")
66        if value in {"yesterday", "昨日", "昨天"}:
67            return (datetime.now() - timedelta(days=1)).strftime("%Y%m%d")
68        return str(value).replace("-", "").replace("/", "")
69 
70    def _clean_symbol(self, symbol: Optional[str]) -> str:
71        if not symbol:
72            return ""
73        return str(symbol).lower().replace("sz", "").replace("sh", "").replace("bj", "")
74 
75    def _market_from_symbol(self, symbol: str) -> str:
76        market = "sh"
77        if symbol.startswith(("0", "3")):
78            market = "sz"
79        elif symbol.startswith(("8", "4")):
80            market = "bj"
81        return market
82 
83    def _filter_records_by_symbol(self, records: list[dict], symbol: str) -> list[dict]:
84        if not symbol:
85            return records
86 
87        key_pool = ["代码", "股票代码", "证券代码", "symbol", "代码简称"]
88        filtered = []
89        for row in records:
90            if not isinstance(row, dict):
91                continue
92            for key in key_pool:
93                val = row.get(key)
94                if val is not None and symbol in str(val):
95                    filtered.append(row)
96                    break
97        return filtered
98 
99    def _call_api_candidates(self, candidates: list[tuple[str, list[dict]]]) -> tuple[Optional[str], Any, str]:
100        errors = []
101 
102        for fn_name, kwargs_list in candidates:
103            func = getattr(self._ak, fn_name, None)
104            if func is None:
105                continue
106 
107            args_pool = kwargs_list or [{}]
108            for kwargs in args_pool:
109                try:
110                    result = func(**kwargs)
111                    return fn_name, result, ""
112                except Exception as exc:
113                    errors.append(f"{fn_name}({kwargs}): {exc}")
114 
115        return None, None, "; ".join(errors) if errors else "no callable api found"
116 
117    def index_spot(self, top_n: int = 300) -> Dict[str, Any]:
118        primary_fn = "stock_zh_index_spot_sina"
119        err = self._ready_or_error(primary_fn)
120        if err:
121            return err
122 
123        try:
124            df = self._ak.stock_zh_index_spot_sina()
125            return self._wrap(primary_fn, items=self._to_records(df, top_n=top_n))
126        except Exception as exc:
127            fallback_fn = "stock_zh_index_spot_em"
128            try:
129                df = self._ak.stock_zh_index_spot_em()
130                return self._wrap(fallback_fn, items=self._to_records(df, top_n=top_n))
131            except Exception as fallback_exc:
132                return self._error(primary_fn, f"sina failed: {exc}; em failed: {fallback_exc}")
133 
134    def stock_kline(
135        self,
136        symbol: str,
137        period: str = "daily",
138        start_date: Optional[str] = None,
139        end_date: Optional[str] = None,
140        top_n: int = 60,
141    ) -> Dict[str, Any]:
142        fn_name = "stock_zh_a_hist"
143        err = self._ready_or_error(fn_name)
144        if err:
145            return err
146 
147        if not start_date:
148            end_dt = datetime.now()
149            if period == "weekly":
150                days = top_n * 7
151            elif period == "monthly":
152                days = top_n * 30
153            else:
154                days = top_n
155            start_dt = end_dt - timedelta(days=days + 50)
156            start = start_dt.strftime("%Y%m%d")
157        else:
158            start = start_date.replace("-", "")
159 
160        end = self._normalize_trade_date(end_date)
161 
162        try:
163            df = self._ak.stock_zh_a_hist(
164                symbol=symbol,
165                period=period,
166                start_date=start,
167                end_date=end,
168                adjust="",
169            )
170            if hasattr(df, "iloc"):
171                df = df.iloc[::-1]
172            return self._wrap(
173                fn_name,
174                symbol=symbol,
175                period=period,
176                start_date=start,
177                end_date=end,
178                items=self._to_records(df, top_n=top_n),
179            )
180        except Exception as exc:
181            return self._error(fn_name, str(exc))
182 
183    def stock_chart(self, symbol: str, period: str = "daily", days: int = 30) -> Dict[str, Any]:
184        """生成股票K线图"""
185        fn_name = "stock_chart"
186        err = self._ready_or_error(fn_name)
187        if err:
188            return err
189 
190        try:
191            import matplotlib
192            matplotlib.use('Agg')
193            import matplotlib.pyplot as plt
194            import matplotlib.font_manager as fm
195            
196            # 设置中文字体
197            font_paths = [
198                '/Library/Fonts/Microsoft/SimHei.ttf',
199                '/Library/Fonts/Microsoft/Microsoft Yahei.ttf',
200                '/Users/molezz/Library/Fonts/msyh.ttf',
201                '/System/Library/Fonts/STHeiti Medium.ttc',
202            ]
203            found_font = None
204            for fp in font_paths:
205                if os.path.exists(fp):
206                    found_font = fp
207                    break
208            if found_font:
209                fm.fontManager.addfont(found_font)
210                prop = fm.FontProperties(fname=found_font)
211                plt.rcParams['font.sans-serif'] = [prop.get_name()]
212                plt.rcParams['axes.unicode_minus'] = False
213            
214            # 计算日期
215            from datetime import datetime, timedelta
216            end_date = datetime.now().strftime("%Y%m%d")
217            start_date = (datetime.now() - timedelta(days=days+30)).strftime("%Y%m%d")
218            
219            # 获取数据
220            df = self._ak.stock_zh_a_hist(symbol=symbol, period=period, start_date=start_date, end_date=end_date)
221            if df is None or len(df) == 0:
222                return self._error(fn_name, "无法获取数据")
223            
224            # 取最近的数据
225            df = df.tail(days)
226            
227            # 获取股票名称
228            name = symbol
229            try:
230                info = self._ak.stock_individual_info_em(symbol=symbol)
231                if info is not None and len(info) > 0:
232                    # 尝试获取"股票简称"
233                    name_row = info[info.get('item', '') == '股票简称']
234                    if len(name_row) > 0:
235                        name = name_row.iloc[0].get('value', symbol)
236                    else:
237                        # 如果没有简称，用代码
238                        name = symbol
239            except:
240                pass
241            
242            # 绘图
243            plt.figure(figsize=(10, 6))
244            plt.plot(df['日期'], df['收盘'], 'b-', linewidth=1.5)
245            plt.title(f'{name}({symbol}) 近期股价走势', fontsize=14)
246            plt.xlabel('日期')
247            plt.ylabel('收盘价 (元)')
248            plt.grid(True, alpha=0.3)
249            plt.xticks(rotation=45)
250            plt.tight_layout()
251            
252            # 保存
253            chart_dir = "/tmp/stock_charts"
254            os.makedirs(chart_dir, exist_ok=True)
255            filepath = f"{chart_dir}/{symbol}.png"
256            plt.savefig(filepath, dpi=120)
257            plt.close()
258            
259            return {
260                "ok": True,
261                "data": {
262                    "symbol": symbol,
263                    "name": name,
264                    "filepath": filepath,
265                    "period": period,
266                    "days": days,
267                },
268                "image_path": filepath
269            }
270        except Exception as exc:
271            return self._error(fn_name, str(exc))
272 
273    def stock_intraday(self, symbol: str, period: Optional[str] = None, top_n: int = 30) -> Dict[str, Any]:
274        fn_name = "stock_intraday"
275        err = self._ready_or_error(fn_name)
276        if err:
277            return err
278 
279        minute_error = None
280        minute_period = period if period in {"1", "5", "15", "30", "60"} else "1"
281 
282        try:
283            df = self._ak.stock_zh_a_minute(symbol=symbol, period=minute_period, adjust="")
284            if hasattr(df, "iloc"):
285                df = df.iloc[::-1]
286            return self._wrap(
287                "stock_zh_a_minute",
288                symbol=symbol,
289                period=minute_period,
290                items=self._to_records(df, top_n=top_n),
291            )
292        except Exception as exc:
293            minute_error = str(exc)
294 
295        try:
296            df = self._ak.stock_intraday_em(symbol=symbol)
297            return self._wrap(
298                "stock_intraday_em",
299                symbol=symbol,
300                period="tick",
301                fallback=minute_error,
302                items=self._to_records(df, top_n=top_n),
303            )
304        except Exception as exc:
305            if minute_error:
306                return self._error(fn_name, f"minute failed: {minute_error}; tick failed: {exc}")
307            return self._error(fn_name, str(exc))
308 
309    def limit_pool(self, date: Optional[str] = None, top_n: int = 50) -> Dict[str, Any]:
310        fn_name = "stock_zt_pool_em"
311        err = self._ready_or_error(fn_name)
312        if err:
313            return err
314 
315        trade_date = self._normalize_trade_date(date)
316 
317        try:
318            up_df = self._ak.stock_zt_pool_em(date=trade_date)
319            up_count = self._data_len(up_df)
320            up_items = self._to_records(up_df, top_n=top_n)
321 
322            down_count = 0
323            down_items: Any = []
324            down_api = None
325            down_errors = []
326 
327            for api_name in ["stock_zt_pool_dtgc_em", "stock_dt_pool_em"]:
328                func = getattr(self._ak, api_name, None)
329                if func is None:
330                    continue
331                try:
332                    down_df = func(date=trade_date)
333                    down_count = self._data_len(down_df)
334                    down_items = self._to_records(down_df, top_n=top_n)
335                    down_api = api_name
336                    break
337                except Exception as exc:
338                    down_errors.append(f"{api_name}: {exc}")
339 
340            payload: Dict[str, Any] = {
341                "date": trade_date,
342                "up_count": up_count,
343                "down_count": down_count,
344                "up_items": up_items,
345                "down_items": down_items,
346                "items": up_items,
347            }
348            if down_api:
349                payload["down_api"] = down_api
350            if down_errors and not down_api:
351                payload["down_error"] = "; ".join(down_errors)
352 
353            return self._wrap(fn_name, **payload)
354        except Exception as exc:
355            return self._error(fn_name, str(exc))
356 
357    def news(self, top_n: int = 10) -> Dict[str, Any]:
358        """财经要闻
359 
360        旧实现依赖 akshare.stock_news_em（东财接口），在部分环境下可能长期返回历史日期。
361        这里改为用 agent-browser 抓取「东方财富财经首页」的最新要闻链接。
362 
363        返回字段尽量与原 formatter 兼容：新闻标题/新闻链接/发布时间/文章来源。
364        """
365 
366        fn_name = "eastmoney_finance_home"
367 
368        # 1) Primary: agent-browser scrape
369        try:
370            import json
371            import subprocess
372 
373            n = max(1, min(int(top_n or 10), 20))
374 
375            # 打开财经首页（复用默认 session，执行很快）
376            subprocess.run(
377                ["agent-browser", "open", "https://finance.eastmoney.com/"],
378                capture_output=True,
379                text=True,
380                timeout=20,
381                check=False,
382            )
383 
384            js = r"""
385(() => {
386  const now = new Date();
387  const pad2 = (x) => String(x).padStart(2, '0');
388  const today = `${now.getFullYear()}-${pad2(now.getMonth()+1)}-${pad2(now.getDate())}`;
389  const ymd = `${now.getFullYear()}${pad2(now.getMonth()+1)}${pad2(now.getDate())}`;
390  const ymdYesterday = (() => {
391    const d = new Date(now.getTime() - 24*3600*1000);
392    return `${d.getFullYear()}${pad2(d.getMonth()+1)}${pad2(d.getDate())}`;
393  })();
394 
395  const links = Array.from(document.querySelectorAll('a[href]'));
396  const items = [];
397  const seen = new Set();
398 
399  for (const a of links) {
400    const href = a.href || '';
401    if (!href.includes('finance.eastmoney.com/a/')) continue;
402    // 排除频道页（/a/cxxxx.html）
403    if (/\/a\/c\w+\.html/.test(href)) continue;
404 
405    // 只保留今天/昨天的文章（避免首页混入更早的深链）
406    const dm = href.match(/\/a\/(\d{8})\d+\.html/);
407    if (!dm) continue;
408    const d8 = dm[1];
409    if (!(d8 === ymd || d8 === ymdYesterday)) continue;
410 
411    const title = (a.textContent || '').replace(/\s+/g, ' ').trim();
412    if (!title || title.length < 6) continue;
413    if (seen.has(href)) continue;
414    seen.add(href);
415 
416    const publish = `${d8.slice(0,4)}-${d8.slice(4,6)}-${d8.slice(6,8)}`;
417 
418    items.push({
419      '新闻标题': title,
420      '新闻链接': href,
421      '发布时间': publish || today,
422      '文章来源': '东方财富网'
423    });
424 
425    if (items.length >= 80) break;
426  }
427 
428  return JSON.stringify(items);
429})();
430"""
431 
432            p = subprocess.run(
433                ["agent-browser", "eval", js],
434                capture_output=True,
435                text=True,
436                timeout=20,
437                check=False,
438            )
439 
440            if p.returncode == 0 and p.stdout:
441                raw = p.stdout.strip()
442                # agent-browser 可能会把 JSON 字符串再包一层引号，这里统一处理
443                # agent-browser 的输出有两种形态：
444                # 1) 直接 JSON 数组：[{...},{...}]
445                # 2) JSON 字符串（外层带引号）："[{...},{...}]"
446                data = json.loads(raw)
447                if isinstance(data, str):
448                    data = json.loads(data)
449 
450                items = (data or [])[:n]
451                return self._wrap(fn_name, items=items)
452 
453            # fallthrough to error
454            err_msg = (p.stderr or p.stdout or "agent-browser eval failed").strip()
455            return self._error(fn_name, err_msg)
456 
457        except Exception as exc:
458            return self._error(fn_name, str(exc))
459 
460    def research_report(self, symbol: str, top_n: int = 10) -> Dict[str, Any]:
461        fn_name = "stock_research_report_em"
462        err = self._ready_or_error(fn_name)
463        if err:
464            return err
465 
466        clean_symbol = self._clean_symbol(symbol)
467        if not clean_symbol:
468            return self._error(fn_name, "symbol is required")
469 
470        try:
471            with redirect_stdout(StringIO()), redirect_stderr(StringIO()):
472                df = self._ak.stock_research_report_em(symbol=clean_symbol)
473            items = self._to_records(df, top_n=max(1, min(top_n, 10)))
474            return self._wrap(fn_name, symbol=clean_symbol, items=items)
475        except Exception as exc:
476            return self._error(fn_name, str(exc))
477 
478    def money_flow(self, symbol: str, top_n: int = 30) -> Dict[str, Any]:
479        fn_name = "stock_individual_fund_flow"
480        err = self._ready_or_error(fn_name)
481        if err:
482            return err
483 
484        clean_symbol = self._clean_symbol(symbol)
485        market = self._market_from_symbol(clean_symbol)
486 
487        try:
488            df = self._ak.stock_individual_fund_flow(stock=clean_symbol, market=market)
489            if hasattr(df, "iloc"):
490                df = df.iloc[::-1]
491            return self._wrap(
492                fn_name,
493                scope="individual",
494                symbol=clean_symbol,
495                market=market,
496                items=self._to_records(df, top_n=top_n),
497            )
498        except Exception as exc:
499            return self._error(fn_name, str(exc))
500 
501    def market_money_flow(self, top_n: int = 20, date: Optional[str] = None) -> Dict[str, Any]:
502        fn_name = "market_money_flow"
503        err = self._ready_or_error(fn_name)
504        if err:
505            return err
506 
507        trade_date = self._normalize_trade_date(date)
508 
509        candidates = [
510            ("stock_market_fund_flow", [{}]),
511            ("stock_hsgt_fund_flow_summary_em", [{}]),
512            ("stock_hsgt_north_net_flow_in_em", [{}]),
513            ("stock_hsgt_hist_em", [{"symbol": "北向资金"}, {"symbol": "沪股通"}, {"symbol": "深股通"}]),
514        ]
515 
516        api_name, df, err_msg = self._call_api_candidates(candidates)
517        if df is None:
518            return self._error(fn_name, err_msg)
519 
520        if hasattr(df, "iloc"):
521            try:
522                df = df.iloc[::-1]
523            except Exception:
524                pass
525 
526        return self._wrap(
527            api_name or fn_name,
528            scope="market",
529            date=trade_date,
530            items=self._to_records(df, top_n=top_n),
531        )
532 
533    def sector_money_flow(self, top_n: int = 20) -> Dict[str, Any]:
534        fn_name = "sector_money_flow"
535        err = self._ready_or_error(fn_name)
536        if err:
537            return err
538 
539        candidates = [
540            (
541                "stock_sector_fund_flow_rank",
542                [
543                    {"indicator": "今日", "sector_type": "行业资金流"},
544                    {"indicator": "5日", "sector_type": "行业资金流"},
545                    {"indicator": "10日", "sector_type": "行业资金流"},
546                    {"symbol": "今日", "sector_type": "行业资金流"},
547                    {"sector_type": "行业资金流"},
548                ],
549            ),
550            ("stock_fund_flow_industry", [{"symbol": "今日"}, {"symbol": "即时"}, {}]),
551            ("stock_sector_fund_flow_summary", [{"sector_type": "行业资金流"}, {}]),
552        ]
553 
554        api_name, df, err_msg = self._call_api_candidates(candidates)
555        if df is None:
556            return self._error(fn_name, err_msg)
557 
558        return self._wrap(
559            api_name or fn_name,
560            scope="sector",
561            items=self._to_records(df, top_n=top_n),
562        )
563 
564    def fundamental(self, symbol: str, top_n: int = 20) -> Dict[str, Any]:
565        fn_name = "fundamental"
566        err = self._ready_or_error(fn_name)
567        if err:
568            return err
569 
570        clean_symbol = self._clean_symbol(symbol)
571 
572        candidates = [
573            (
574                "stock_financial_abstract_ths",
575                [
576                    {"symbol": clean_symbol, "indicator": "按报告期"},
577                    {"symbol": clean_symbol, "indicator": "按单季度"},
578                    {"symbol": clean_symbol},
579                    {"stock": clean_symbol, "indicator": "按报告期"},
580                    {"stock": clean_symbol},
581                ],
582            ),
583            (
584                "stock_financial_analysis_indicator",
585                [
586                    {"symbol": clean_symbol},
587                    {"stock": clean_symbol},
588                ],
589            ),
590        ]
591 
592        api_name, df, err_msg = self._call_api_candidates(candidates)
593        if df is None:
594            return self._error(fn_name, err_msg)
595 
596        if hasattr(df, "iloc"):
597            try:
598                df = df.iloc[::-1]
599            except Exception:
600                pass
601 
602        items = self._to_records(df, top_n=top_n)
603        latest = items[0] if isinstance(items, list) and items else {}
604 
605        return self._wrap(
606            api_name or fn_name,
607            scope="fundamental",
608            symbol=clean_symbol,
609            latest=latest,
610            items=items,
611        )
612 
613    def stock_overview(self, symbol: str) -> Dict[str, Any]:
614        fn_name = "stock_overview"
615        clean_symbol = self._clean_symbol(symbol)
616 
617        if not clean_symbol:
618            return self._error(fn_name, "symbol is required")
619 
620        sections: Dict[str, Any] = {
621            "realtime": {"ok": False, "error": "not called"},
622            "money_flow": {"ok": False, "error": "not called"},
623            "fundamental": {"ok": False, "error": "not called"},
624            "limit_stats": {"ok": False, "error": "not called"},
625            "research_report": {"ok": False, "error": "not called"},
626        }
627 
628        # 1) 实时行情（优先使用分时最新）
629        try:
630            rt_res = self.stock_intraday(symbol=clean_symbol, period="1", top_n=1)
631            if rt_res.get("ok"):
632                rt_items = rt_res.get("data", {}).get("items", [])
633                latest = rt_items[0] if isinstance(rt_items, list) and rt_items else {}
634                sections["realtime"] = {
635                    "ok": True,
636                    "api": rt_res.get("api"),
637                    "latest": latest,
638                }
639            else:
640                sections["realtime"] = {
641                    "ok": False,
642                    "api": rt_res.get("api"),
643                    "error": rt_res.get("error", "unknown error"),
644                }
645        except Exception as exc:
646            sections["realtime"] = {"ok": False, "error": str(exc)}
647 
648        # 2) 个股资金流
649        try:
650            flow_res = self.money_flow(symbol=clean_symbol, top_n=10)
651            if flow_res.get("ok"):
652                flow_data = flow_res.get("data", {})
653                flow_items = flow_data.get("items", [])
654                sections["money_flow"] = {
655                    "ok": True,
656                    "api": flow_res.get("api"),
657                    "latest": flow_items[0] if isinstance(flow_items, list) and flow_items else {},
658                    "items": flow_items,
659                }
660            else:
661                sections["money_flow"] = {
662                    "ok": False,
663                    "api": flow_res.get("api"),
664                    "error": flow_res.get("error", "unknown error"),
665                }
666        except Exception as exc:
667            sections["money_flow"] = {"ok": False, "error": str(exc)}
668 
669        # 3) 基本面摘要
670        try:
671            fundamental_res = self.fundamental(symbol=clean_symbol, top_n=10)
672            if fundamental_res.get("ok"):
673                fundamental_data = fundamental_res.get("data", {})
674                sections["fundamental"] = {
675                    "ok": True,
676                    "api": fundamental_res.get("api"),
677                    "latest": fundamental_data.get("latest") or {},
678                    "items": fundamental_data.get("items") or [],
679                }
680            else:
681                sections["fundamental"] = {
682                    "ok": False,
683                    "api": fundamental_res.get("api"),
684                    "error": fundamental_res.get("error", "unknown error"),
685                }
686        except Exception as exc:
687            sections["fundamental"] = {"ok": False, "error": str(exc)}
688 
689        # 4) 近期涨跌停（从近10日池中统计该股出现次数）
690        limit_up_count = 0
691        limit_down_count = 0
692        last_date = None
693        limit_errors = []
694 
695        code_keys = ["代码", "股票代码", "证券代码", "symbol"]
696        name_keys = ["名称", "股票简称", "证券简称", "简称"]
697 
698        for offset in range(0, 10):
699            trade_date = (datetime.now() - timedelta(days=offset)).strftime("%Y%m%d")
700            try:
701                limit_res = self.limit_pool(date=trade_date, top_n=300)
702                if not limit_res.get("ok"):
703                    limit_errors.append(f"{trade_date}: {limit_res.get('error', 'unknown error')}")
704                    continue
705 
706                payload = limit_res.get("data", {})
707                up_items = payload.get("up_items") or payload.get("items") or []
708                down_items = payload.get("down_items") or []
709                if last_date is None:
710                    last_date = payload.get("date") or trade_date
711 
712                def _is_target(row: Any) -> bool:
713                    if not isinstance(row, dict):
714                        return False
715                    for key in code_keys:
716                        value = row.get(key)
717                        if value is not None and clean_symbol == self._clean_symbol(str(value)):
718                            return True
719                    for key in name_keys:
720                        value = row.get(key)
721                        if value is not None and str(value) in str(symbol):
722                            return True
723                    return False
724 
725                limit_up_count += sum(1 for row in up_items if _is_target(row))
726                limit_down_count += sum(1 for row in down_items if _is_target(row))
727            except Exception as exc:
728                limit_errors.append(f"{trade_date}: {exc}")
729 
730        sections["limit_stats"] = {
731            "ok": True,
732            "days": 10,
733            "date": last_date,
734            "up_count": limit_up_count,
735            "down_count": limit_down_count,
736            "error": "; ".join(limit_errors[:3]) if limit_errors else None,
737        }
738 
739        # 5) 研报
740        try:
741            report_res = self.research_report(symbol=clean_symbol, top_n=3)
742            if report_res.get("ok"):
743                report_data = report_res.get("data", {})
744                sections["research_report"] = {
745                    "ok": True,
746                    "api": report_res.get("api"),
747                    "items": report_data.get("items", [])[:3],
748                }
749            else:
750                sections["research_report"] = {
751                    "ok": False,
752                    "api": report_res.get("api"),
753                    "error": report_res.get("error", "unknown error"),
754                }
755        except Exception as exc:
756            sections["research_report"] = {"ok": False, "error": str(exc)}
757 
758        has_success = any(section.get("ok") for section in sections.values())
759        if not has_success:
760            combined_error = "; ".join(
761                str(section.get("error"))
762                for section in sections.values()
763                if section.get("error")
764            )
765            return self._error(fn_name, combined_error or "all sub-apis failed")
766 
767        return self._wrap(
768            fn_name,
769            symbol=clean_symbol,
770            realtime=sections["realtime"],
771            money_flow=sections["money_flow"],
772            fundamental=sections["fundamental"],
773            limit_stats=sections["limit_stats"],
774            research_report=sections["research_report"],
775        )
776 
777    def stock_pick(self, top_n: int = 5, sector: str = None) -> Dict[str, Any]:
778        import warnings
779        fn_name = "stock_pick"
780        err = self._ready_or_error(fn_name)
781        if err:
782            return err
783 
784        def pick(item: dict, keys: list, default: Any = None) -> Any:
785            for key in keys:
786                value = item.get(key)
787                if value not in (None, ""):
788                    return value
789            return default
790 
791        def normalize_code(value: Any) -> str:
792            if value is None:
793                return ""
794            text = str(value).strip().upper()
795            if not text:
796                return ""
797            text = text.replace("SH", "").replace("SZ", "").replace("BJ", "")
798            digits = "".join(ch for ch in text if ch.isdigit())
799            if len(digits) >= 6:
800                return digits[:6]
801            return text
802 
803        # 板块关键词映射
804        sector_keywords = {
805            "半导体": ["半导体", "芯片", "集成电路"],
806            "电子": ["电子", "科技", "计算机"],
807            "汽车": ["汽车", "新能源车", "整车", "汽配"],
808            "医药生物": ["医药", "医疗器械", "中药", "生物医药", "医疗", "医药生物"],
809            "医药": ["医药", "医疗器械", "中药", "生物医药", "医疗", "医药生物"],
810            "光伏": ["光伏", "光伏发电", "光伏设备"],
811            "锂电池": ["锂电池", "锂电", "电池", "动力电池"],
812            "新能源": ["新能源", "储能", "电动车", "电动汽车"],
813            "银行": ["银行", "银行股"],
814            "保险": ["保险", "保险股"],
815            "证券": ["证券", "券商"],
816            "金融": ["金融", "银行", "保险", "证券"],
817            "房地产": ["房地产", "地产", "物业"],
818            "地产": ["房地产", "地产", "物业"],
819            "电力": ["电力", "电力股", "发电"],
820            "传媒": ["传媒", "影视", "游戏"],
821            "军工": ["军工", "航天", "航空", "船舶", "国防"],
822            "软件": ["软件", "互联网", "计算机", "IT", "软件开发"],
823            "食品": ["食品", "零食", "食品加工"],
824            "饮料": ["饮料", "饮品"],
825            "白酒": ["白酒", "酒", "白酒股"],
826            "家电": ["家电", "白色家电", "冰洗"],
827            "纺织": ["纺织", "纺织服装", "服装"],
828        }
829 
830        # 板块关键词映射到接口参数（使用 akshare 实际支持的名称）
831        sector_map = {
832            # 常用板块
833            "半导体": "半导体",
834            "电子": "电子",
835            "汽车": "汽车",
836            "医药生物": "医药生物",
837            "医药": "医药生物",
838            "银行": "银行",
839            "保险": "保险",
840            "证券": "证券",
841            "房地产": "房地产",
842            "锂电池": "锂电池",
843            "电池": "电池",
844            "光伏": "光伏设备",
845            "光伏设备": "光伏设备",
846            "电力": "电力",
847            "传媒": "传媒",
848            "军工": "军工",
849            "软件": "软件开发",
850            "食品": "食品",
851            "饮料": "饮料",
852            "白酒": "白酒",
853            "家电": "家电",
854            "纺织": "纺织",
855        }
856 
857        target_sector = None
858        target_symbol = None
859        if sector:
860            sector_lower = sector.lower()
861            for key, keywords in sector_keywords.items():
862                if any(k in sector_lower for k in keywords):
863                    target_sector = key
864                    target_symbol = sector_map.get(key, key)
865                    break
866 
867        # 1. 如果指定了板块，获取板块成分股
868        sector_stocks = []
869        if target_sector and target_symbol:
870            try:
871                with warnings.catch_warnings():
872                    warnings.simplefilter("ignore")
873                    df = self._ak.stock_board_industry_cons_em(symbol=target_symbol)
874                if hasattr(df, 'to_dict'):
875                    for row in df.to_dict(orient='records'):
876                        if not isinstance(row, dict):
877                            continue
878                        code = normalize_code(pick(row, ["代码", "股票代码"]))
879                        name = pick(row, ["名称", "股票名称"], "")
880                        pct = pick(row, ["涨跌幅"])
881                        if code:
882                            pct_num = _safe_float_local(pct)
883                            sector_stocks.append({
884                                "code": code,
885                                "name": str(name) if name else code,
886                                "pct": pct_num if pct_num else 0,
887                            })
888            except Exception as e:
889                pass
890 
891        # 如果成功获取到板块成分股，直接用这些数据
892        if sector_stocks:
893            sector_stocks.sort(key=lambda x: x.get("pct", 0), reverse=True)
894            top_candidates = sector_stocks[:top_n]
895        else:
896            # 2. 获取热门股票（涨跌幅排行）
897            try:
898                import warnings
899                with warnings.catch_warnings():
900                    warnings.simplefilter("ignore")
901                    hot_df = self._ak.stock_hot_rank_em()
902            except Exception as e:
903                return self._error(fn_name, f"热门股票获取失败: {e}")
904 
905            hot_items = []
906            if hasattr(hot_df, 'to_dict'):
907                records = hot_df.to_dict(orient='records')
908                for row in records:
909                    if not isinstance(row, dict):
910                        continue
911                    code = normalize_code(pick(row, ["代码", "股票代码", "证券代码", "symbol"]))
912                    name = pick(row, ["股票名称", "名称", "简称", "name"], "")
913                    pct = pick(row, ["涨跌幅", "涨跌幅%"])
914                    if code:
915                        pct_num = _safe_float_local(pct)
916                        hot_items.append({
917                            "code": code,
918                            "name": str(name) if name else code,
919                            "pct": pct_num if pct_num else 0,
920                        })
921 
922            if not hot_items:
923                return self._error(fn_name, "热门股票数据为空")
924 
925            hot_items.sort(key=lambda x: x.get("pct", 0), reverse=True)
926            top_candidates = hot_items[:10]
927 
928        # 获取行业资金流
929        try:
930            with warnings.catch_warnings():
931                warnings.simplefilter("ignore")
932                sector_res = self.sector_money_flow(top_n=15)
933        except:
934            sector_res = {"ok": False}
935 
936        hot_industries = set()
937        if sector_res.get("ok"):
938            sector_items = sector_res.get("data", {}).get("items", [])
939            for row in sector_items:
940                if not isinstance(row, dict):
941                    continue
942                name = pick(row, ["名称", "行业"])
943                inflow = _safe_float_local(pick(row, ["今日主力净流入-净额", "主力净流入"]))
944                if name and inflow and inflow > 0:
945                    hot_industries.add(str(name).strip())
946 
947        # 3. 简化：只取研报数据（不做个股详细查询）
948        report_map = {}
949        try:
950            with warnings.catch_warnings():
951                warnings.simplefilter("ignore")
952                report_df = self._ak.stock_research_report_em()
953            if hasattr(report_df, 'to_dict'):
954                report_records = report_df.to_dict(orient='records')[:50]
955                for row in report_records:
956                    if not isinstance(row, dict):
957                        continue
958                    code = normalize_code(pick(row, ["股票代码", "代码"]))
959                    rating = str(pick(row, ["东财评级", "评级"], ""))
960                    if "买入" in rating and code not in report_map:
961                        report_map[code] = {
962                            "org": pick(row, ["机构"], "机构"),
963                            "rating": rating,
964                            "title": str(pick(row, ["报告名称"], ""))[:20],
965                        }
966        except:
967            pass
968 
969        # 4. 组装推荐结果
970        selected = []
971        for row in top_candidates:
972            code = row["code"]
973            name = row["name"]
974            pct = row["pct"]
975            
976            report = report_map.get(code, {})
977            
978            selected.append({
979                "name": name,
980                "code": code,
981                "pct": pct,
982                "report_org": report.get("org", ""),
983                "report_rating": report.get("rating", ""),
984                "report_title": report.get("title", ""),
985            })
986 
987        return self._wrap(
988            fn_name,
989            items=selected[:top_n],
990            count=len(selected),
991        )
992 
993    def margin_lhb(self, symbol: Optional[str] = None, date: Optional[str] = None, top_n: int = 10) -> Dict[str, Any]:
994        fn_name = "margin_lhb"
995        err = self._ready_or_error(fn_name)
996        if err:
997            return err
998 
999        clean_symbol = self._clean_symbol(symbol)
1000        trade_date = self._normalize_trade_date(date)
1001 
1002        margin_candidates = [
1003            (
1004                "stock_margin_detail",
1005                [
1006                    {"date": trade_date, "symbol": clean_symbol},
1007                    {"date": trade_date, "stock": clean_symbol},
1008                    {"date": trade_date, "code": clean_symbol},
1009                    {"date": trade_date},
1010                ],
1011            ),
1012            ("stock_margin_detail_em", [{"date": trade_date}, {"trade_date": trade_date}, {}]),
1013            ("stock_margin_underlying_info_szse", [{}]),
1014            ("stock_margin_underlying_info_sse", [{}]),
1015        ]
1016 
1017        margin_api, margin_df, margin_err = self._call_api_candidates(margin_candidates)
1018        margin_items: list[dict] = []
1019        if margin_df is not None:
1020            margin_items = self._to_records(margin_df, top_n=0)
1021            if isinstance(margin_items, list):
1022                margin_items = [item for item in margin_items if isinstance(item, dict)]
1023                margin_items = self._filter_records_by_symbol(margin_items, clean_symbol)
1024                margin_items = margin_items[:top_n]
1025            else:
1026                margin_items = []
1027 
1028        lhb_candidates = [
1029            (
1030                "stock_lhb_detail_em",
1031                [
1032                    {"start_date": trade_date, "end_date": trade_date},
1033                    {"date": trade_date},
1034                    {},
1035                ],
1036            ),
1037            ("stock_lhb_ggtj_sina", [{"symbol": "5"}, {"symbol": "10"}, {}]),
1038        ]
1039 
1040        lhb_api, lhb_df, lhb_err = self._call_api_candidates(lhb_candidates)
1041        lhb_items: list[dict] = []
1042        if lhb_df is not None:
1043            lhb_items = self._to_records(lhb_df, top_n=0)
1044            if isinstance(lhb_items, list):
1045                lhb_items = [item for item in lhb_items if isinstance(item, dict)]
1046                lhb_items = self._filter_records_by_symbol(lhb_items, clean_symbol)
1047                lhb_items = lhb_items[:top_n]
1048            else:
1049                lhb_items = []
1050 
1051        if margin_df is None and lhb_df is None:
1052            return self._error(fn_name, f"margin failed: {margin_err}; lhb failed: {lhb_err}")
1053 
1054        return self._wrap(
1055            fn_name,
1056            scope="margin_lhb",
1057            symbol=clean_symbol,
1058            date=trade_date,
1059            margin_api=margin_api,
1060            lhb_api=lhb_api,
1061            margin_items=margin_items,
1062            lhb_items=lhb_items,
1063            margin_error=margin_err if margin_df is None else None,
1064            lhb_error=lhb_err if lhb_df is None else None,
1065        )
1066 
1067    def sector_analysis(self, sector_type: str = "industry", top_n: int = 10) -> Dict[str, Any]:
1068        fn_name = "stock_sector_name_code"
1069        err = self._ready_or_error(fn_name)
1070        if err:
1071            return err
1072 
1073        normalized = "概念" if sector_type in {"concept", "概念"} else "行业"
1074        spot_indicator = "概念" if normalized == "概念" else "新浪行业"
1075        candidates = [
1076            ("stock_sector_name_code", [{"indicator": "今日涨跌幅", "sector_type": normalized}]),
1077            ("stock_sector_name_code", [{"sector_type": normalized}]),
1078            ("stock_sector_spot", [{"indicator": spot_indicator}]),
1079        ]
1080 
1081        api_name, df, err_msg = self._call_api_candidates(candidates)
1082        if df is None:
1083            return self._error(fn_name, err_msg)
1084 
1085        records = self._to_records(df, top_n=0)
1086        if isinstance(records, list):
1087            records = [item for item in records if isinstance(item, dict)]
1088            records.sort(
1089                key=lambda row: _safe_float_local(
1090                    row.get("涨跌幅")
1091                    or row.get("今日涨跌幅")
1092                    or row.get("涨跌幅%")
1093                    or row.get("涨跌")
1094                )
1095                or -9999,
1096                reverse=True,
1097            )
1098            top_gain = records[:top_n]
1099            top_drop = sorted(
1100                records,
1101                key=lambda row: _safe_float_local(
1102                    row.get("涨跌幅")
1103                    or row.get("今日涨跌幅")
1104                    or row.get("涨跌幅%")
1105                    or row.get("涨跌")
1106                )
1107                or 9999,
1108            )[:top_n]
1109        else:
1110            top_gain = []
1111            top_drop = []
1112 
1113        return self._wrap(
1114            api_name or fn_name,
1115            scope="sector_analysis",
1116            sector_type="concept" if normalized == "概念" else "industry",
1117            top_gain=top_gain,
1118            top_drop=top_drop,
1119            items=top_gain,
1120        )
1121 
1122    def fund_bond(self, scope: str = "fund", symbol: Optional[str] = None, top_n: int = 10) -> Dict[str, Any]:
1123        fn_name = "fund_bond"
1124        err = self._ready_or_error(fn_name)
1125        if err:
1126            return err
1127 
1128        normalized_scope = "bond" if scope in {"bond", "convertible", "cb"} else "fund"
1129 
1130        if normalized_scope == "fund":
1131            clean_symbol = self._clean_symbol(symbol)
1132            default_symbol = clean_symbol or "159915"
1133            candidates = [
1134                (
1135                    "fund_etf_hist_em",
1136                    [
1137                        {
1138                            "symbol": default_symbol,
1139                            "period": "daily",
1140                            "start_date": (datetime.now() - timedelta(days=90)).strftime("%Y%m%d"),
1141                            "end_date": datetime.now().strftime("%Y%m%d"),
1142                            "adjust": "",
1143                        }
1144                    ],
1145                ),
1146                ("fund_etf_spot_em", [{}]),
1147                ("fund_open_fund_daily_em", [{}]),
1148            ]
1149            api_name, df, err_msg = self._call_api_candidates(candidates)
1150            if df is None:
1151                return self._error(fn_name, err_msg)
1152 
1153            records = self._to_records(df, top_n=0)
1154            if isinstance(records, list):
1155                records = [item for item in records if isinstance(item, dict)]
1156                if clean_symbol:
1157                    records = self._filter_records_by_symbol(records, clean_symbol) or records
1158                for item in records:
1159                    if "代码" not in item:
1160                        item["代码"] = default_symbol
1161                if records and "日期" in records[0]:
1162                    try:
1163                        records = sorted(records, key=lambda r: r.get("日期") or "", reverse=True)
1164                    except Exception:
1165                        pass
1166                records = records[:top_n]
1167            else:
1168                records = []
1169 
1170            return self._wrap(
1171                api_name or fn_name,
1172                scope="fund",
1173                symbol=default_symbol,
1174                items=records,
1175            )
1176 
1177        candidates = [
1178            ("bond_zh_hs_cov_spot", [{}]),
1179            ("bond_zh_hs_cov_daily", [{"symbol": symbol or "sh113527"}]),
1180        ]
1181 
1182        api_name, df, err_msg = self._call_api_candidates(candidates)
1183        if df is None:
1184            return self._error(fn_name, err_msg)
1185 
1186        records = self._to_records(df, top_n=0)
1187        if isinstance(records, list):
1188            records = [item for item in records if isinstance(item, dict)]
1189            if symbol:
1190                records = self._filter_records_by_symbol(records, str(symbol)) or records
1191            records = records[:top_n]
1192        else:
1193            records = []
1194 
1195        return self._wrap(
1196            api_name or fn_name,
1197            scope="bond",
1198            symbol=symbol,
1199            items=records,
1200        )
1201 
1202    def hk_us_market(self, market: str = "hk", top_n: int = 10, symbol: Optional[str] = None) -> Dict[str, Any]:
1203        fn_name = "hk_us_market"
1204        err = self._ready_or_error(fn_name)
1205        if err:
1206            return err
1207 
1208        normalized_market = "us" if market in {"us", "美股", "usa"} else "hk"
1209        if normalized_market == "hk":
1210            candidates = [("stock_hk_spot_em", [{}])]
1211        else:
1212            candidates = [("stock_us_spot_em", [{}])]
1213 
1214        api_name, df, err_msg = self._call_api_candidates(candidates)
1215        if df is None:
1216            return self._error(fn_name, err_msg)
1217 
1218        records = self._to_records(df, top_n=0)
1219        if isinstance(records, list):
1220            records = [item for item in records if isinstance(item, dict)]
1221            if symbol:
1222                records = self._filter_records_by_symbol(records, str(symbol)) or records
1223            records = records[:top_n]
1224        else:
1225            records = []
1226 
1227        return self._wrap(
1228            api_name or fn_name,
1229            scope="hk_us_market",
1230            market=normalized_market,
1231            items=records,
1232        )
1233 
1234    def derivatives(self, scope: str = "futures", symbol: Optional[str] = None, top_n: int = 10) -> Dict[str, Any]:
1235        fn_name = "derivatives"
1236        err = self._ready_or_error(fn_name)
1237        if err:
1238            return err
1239 
1240        normalized_scope = "options" if scope in {"option", "options", "期权"} else "futures"
1241 
1242        if normalized_scope == "futures":
1243            candidates = [
1244                ("futures_display_main_sina", [{}]),
1245                ("match_main_contract", [{"symbol": "cffex"}]),
1246                ("futures_main_sina", [{"symbol": "IF0"}, {"symbol": "IH0"}, {"symbol": "IC0"}]),
1247            ]
1248 
1249            api_name, df, err_msg = self._call_api_candidates(candidates)
1250            if df is None:
1251                return self._error(fn_name, err_msg)
1252 
1253            records = self._to_records(df, top_n=0)
1254            if isinstance(records, list):
1255                records = [item for item in records if isinstance(item, dict)]
1256                if symbol:
1257                    records = self._filter_records_by_symbol(records, str(symbol)) or records
1258                records = records[:top_n]
1259            else:
1260                records = []
1261 
1262            return self._wrap(
1263                api_name or fn_name,
1264                scope="futures",
1265                symbol=symbol,
1266                items=records,
1267            )
1268 
1269        candidates = [
1270            ("option_current_em", [{}]),
1271            ("option_cffex_hs300_spot_sina", [{}]),
1272            ("option_finance_board", [{"symbol": "华夏上证50ETF期权"}, {}]),
1273        ]
1274 
1275        api_name, df, err_msg = self._call_api_candidates(candidates)
1276        if df is None:
1277            return self._error(fn_name, err_msg)
1278 
1279        records = self._to_records(df, top_n=0)
1280        if isinstance(records, list):
1281            records = [item for item in records if isinstance(item, dict)]
1282            if symbol:
1283                records = self._filter_records_by_symbol(records, str(symbol)) or records
1284            records = records[:top_n]
1285        else:
1286            records = []
1287 
1288        return self._wrap(
1289            api_name or fn_name,
1290            scope="options",
1291            symbol=symbol,
1292            items=records,
1293        )
1294 
1295 
1296def _safe_float_local(value: Any) -> Optional[float]:
1297    if value is None:
1298        return None
1299    if isinstance(value, str):
1300        value = value.replace(",", "").replace("%", "").strip()
1301    try:
1302        return float(value)
1303    except Exception:
1304        return None
1305