"""Function Calling 工具定义和执行器 让 AI 可以主动调用现有 API 获取市场数据： - get_market_dashboard: 市场看板 - get_theme_history: 题材热点历史 - get_active_core_stocks: 核心股追踪 - get_stock_quote: 个股实时行情 - get_fund_flow: 资金流向 """ import json from services.ai_config import AI_API_BASE, AI_API_KEY, AI_MODEL, AI_MAX_TOKENS, AI_TEMPERATURE TOOLS = [ { "type": "function", "function": { "name": "get_market_dashboard", "description": "获取A股市场看板数据，包含主要指数行情、全市场涨跌统计（涨跌家数/涨停/跌停/炸板率/成交额）、市场温度评分与竞价信号、行业强度榜、概念热度、板块主力资金流、两融余额、海外主要指数与国内期货主力合约（globalMarkets字段）、完整连板梯队（limitLadder字段，含涨停原因与封单金额）、事件情报（热门股/龙虎榜/飙升/异动）", "parameters": {"type": "object", "properties": {}, "required": []} } }, { "type": "function", "function": { "name": "get_theme_history", "description": "获取指定交易日的题材涨幅排行前20", "parameters": { "type": "object", "properties": { "date": {"type": "string", "description": "交易日 YYYY-MM-DD"} }, "required": ["date"] } } }, { "type": "function", "function": { "name": "get_active_core_stocks", "description": "获取活跃核心股列表，包含最近10日涨幅矩阵和所属题材", "parameters": {"type": "object", "properties": {}, "required": []} } }, { "type": "function", "function": { "name": "get_consecutive_core_stocks", "description": "获取连续上榜的核心股列表（近3日及以上连续上榜），用于识别持续活跃的热点股", "parameters": { "type": "object", "properties": { "min_days": {"type": "integer", "description": "最小连续天数，默认3"} }, "required": [] } } }, { "type": "function", "function": { "name": "get_stock_quote", "description": "获取单只股票实时行情（价格、涨跌幅、成交量、换手率等）", "parameters": { "type": "object", "properties": { "code": {"type": "string", "description": "6位股票代码，如 600519"} }, "required": ["code"] } } }, { "type": "function", "function": { "name": "get_fund_flow", "description": "获取个股资金流向数据（主力净流入、超大单/大单/中单/小单流入流出）", "parameters": { "type": "object", "properties": { "code": {"type": "string", "description": "6位股票代码"}, "name": {"type": "string", "description": "股票名称"}, "days": {"type": "integer", "description": "获取天数，默认30"} }, "required": ["code", "name"] } } }, { "type": "function", "function": { "name": "get_news", "description": "获取最近的财经快讯（新浪7x24，含宏观/行业/公司/海外动态），用于重要消息面梳理", "parameters": { "type": "object", "properties": { "limit": {"type": "integer", "description": "获取条数，默认30，最大50"} }, "required": [] } } } ] async def execute_tool(tool_name: str, arguments: dict) -> str: """执行工具调用，返回 JSON 字符串""" try: if tool_name == "get_market_dashboard": return await _get_market_dashboard() elif tool_name == "get_theme_history": return await _get_theme_history(arguments.get("date", "")) elif tool_name == "get_active_core_stocks": return await _get_active_core_stocks() elif tool_name == "get_consecutive_core_stocks": return await _get_consecutive_core_stocks(arguments.get("min_days", 3)) elif tool_name == "get_stock_quote": return await _get_stock_quote(arguments.get("code", "")) elif tool_name == "get_fund_flow": return await _get_fund_flow( arguments.get("code", ""), arguments.get("name", ""), arguments.get("days", 30) ) elif tool_name == "get_news": return await _get_news(arguments.get("limit", 30)) else: return json.dumps({"error": f"未知工具: {tool_name}"}) except Exception as e: return json.dumps({"error": str(e)}) async def _get_news(limit: int = 30) -> str: """获取财经快讯""" from services.market_extra import fetch_news news = await fetch_news(limit) if not news: return json.dumps({"error": "快讯获取失败或暂无数据"}, ensure_ascii=False) return json.dumps({"count": len(news), "items": news}, ensure_ascii=False) async def _get_market_dashboard() -> str: """获取市场看板数据""" from routes.market_dashboard import _build_dashboard data = await _build_dashboard() return json.dumps(data, ensure_ascii=False, default=str) async def _get_theme_history(date: str) -> str: """获取指定交易日题材涨幅前20""" from database import get_connection conn = get_connection() try: rows = conn.execute( "SELECT * FROM daily_top_themes WHERE trade_date = ? ORDER BY rank ASC", (date,) ).fetchall() items = [dict(r) for r in rows] return json.dumps({"date": date, "items": items}, ensure_ascii=False) finally: conn.close() async def _get_active_core_stocks() -> str: """获取活跃核心股列表""" from database import get_connection from datetime import date as date_cls conn = get_connection() try: # 最近10个有数据的交易日 rows = conn.execute( "SELECT DISTINCT trade_date FROM daily_core_stocks ORDER BY trade_date DESC LIMIT 10" ).fetchall() dates = [r["trade_date"] for r in reversed(rows)] if not dates: return json.dumps({"dates": [], "stocks": []}, ensure_ascii=False) placeholders = ",".join("?" * len(dates)) rows = conn.execute( f"""SELECT trade_date, stock_code, stock_name, f3, cover_count FROM daily_core_stocks WHERE trade_date IN ({placeholders}) ORDER BY trade_date DESC, rank ASC""", dates, ).fetchall() stock_days = {} for r in rows: code = r["stock_code"] s = stock_days.setdefault(code, { "stockCode": code, "stockName": r["stock_name"], "coverCount": r["cover_count"], "dailyGains": {}, "appearCount": 0, "lastAppear": None, }) s["dailyGains"][r["trade_date"]] = r["f3"] s["appearCount"] += 1 if s["lastAppear"] is None or r["trade_date"] > s["lastAppear"]: s["lastAppear"] = r["trade_date"] # 只保留最近10日中出现次数最多的前25只（全量可达145KB，会把上下文撑爆） stocks = sorted(stock_days.values(), key=lambda x: -x["appearCount"])[:25] keep_codes = {s["stockCode"] for s in stocks} themes_rows = conn.execute( f"""SELECT stock_code, theme_code, theme_name FROM daily_core_stock_themes WHERE trade_date IN ({placeholders})""", dates, ).fetchall() themes_by_stock = {} for t in themes_rows: if t["stock_code"] not in keep_codes: continue per = themes_by_stock.setdefault(t["stock_code"], {}) per.setdefault(t["theme_code"], {"theme_code": t["theme_code"], "theme_name": t["theme_name"]}) for s in stocks: s["themes"] = list(themes_by_stock.get(s["stockCode"], {}).values())[:5] latest = dates[-1] if dates else None for s in stocks: if s.get("lastAppear") and latest: try: d1 = date_cls.fromisoformat(latest) d2 = date_cls.fromisoformat(s["lastAppear"]) s["daysSinceLastAppear"] = (d1 - d2).days except ValueError: s["daysSinceLastAppear"] = 0 else: s["daysSinceLastAppear"] = 0 return json.dumps({"dates": dates, "stocks": stocks}, ensure_ascii=False) finally: conn.close() async def _get_consecutive_core_stocks(min_days: int = 3) -> str: """获取连续上榜的核心股列表""" from database import get_connection conn = get_connection() try: # 最近30个有数据的交易日 rows = conn.execute( "SELECT DISTINCT trade_date FROM daily_core_stocks ORDER BY trade_date DESC LIMIT 30" ).fetchall() all_dates = [r["trade_date"] for r in reversed(rows)] if not all_dates: return json.dumps({"dates": [], "stocks": []}, ensure_ascii=False) # 查询所有上榜记录 placeholders = ",".join("?" * len(all_dates)) rows = conn.execute( f"""SELECT trade_date, stock_code, stock_name, f3, cover_count FROM daily_core_stocks WHERE trade_date IN ({placeholders}) ORDER BY stock_code, trade_date""", all_dates, ).fetchall() # 按股票分组，记录上榜日期 stock_dates: dict[str, dict] = {} for r in rows: code = r["stock_code"] s = stock_dates.setdefault(code, { "stockCode": code, "stockName": r["stock_name"], "dates": [], "dailyGains": {}, }) s["dates"].append(r["trade_date"]) s["dailyGains"][r["trade_date"]] = r["f3"] # 计算连续上榜天数（从最新交易日往回数） results = [] for code, s in stock_dates.items(): dates_list = sorted(s["dates"]) # 从最新的日期往回数连续天数 consecutive = 0 last_n = [] for i in range(len(dates_list) - 1, -1, -1): if i == len(dates_list) - 1: consecutive = 1 last_n.append(dates_list[i]) else: # 检查是否连续（允许间隔1天非交易日） prev_idx = all_dates.index(dates_list[i + 1]) if dates_list[i + 1] in all_dates else -1 curr_idx = all_dates.index(dates_list[i]) if dates_list[i] in all_dates else -1 if prev_idx - curr_idx <= 2: # 允许间隔1天 consecutive += 1 last_n.append(dates_list[i]) else: break if consecutive >= min_days: results.append({ "stockCode": code, "stockName": s["stockName"], "consecutiveDays": consecutive, "recentDates": last_n, "dailyGains": {d: s["dailyGains"].get(d) for d in last_n}, }) # 按连续天数降序排列 results.sort(key=lambda x: -x["consecutiveDays"]) return json.dumps({ "minDays": min_days, "stocks": results[:30], }, ensure_ascii=False) finally: conn.close() async def _get_stock_quote(code: str) -> str: """获取个股实时行情""" from services.tencent import fetch_quote data = await fetch_quote(code) if not data: return json.dumps({"error": f"未找到股票 {code} 的数据"}) return json.dumps(data, ensure_ascii=False) async def _get_fund_flow(code: str, name: str, days: int) -> str: """获取个股资金流向""" from services.tencent import fetch_quote, get_market_prefix from database import get_connection # 先尝试从本地数据库获取历史资金流向 conn = get_connection() try: rows = conn.execute( "SELECT * FROM cache WHERE key LIKE ? AND expires_at > datetime('now','localtime')", (f"fund_flow:{code}%",) ).fetchall() if rows: return rows[0]["value"] finally: conn.close() # 降级：通过腾讯获取基础行情数据 quote = await fetch_quote(code) if not quote: return json.dumps({"error": f"未找到股票 {code} 的数据"}) return json.dumps({ "code": code, "name": name or quote.get("name", ""), "currentPrice": quote.get("currentPrice", 0), "changePercent": quote.get("changePercent", 0), "volume": quote.get("volume", 0), "amount": quote.get("amount", 0), "note": "资金流向详细数据需通过东方财富API获取" }, ensure_ascii=False)