# -*- coding: utf-8 -*- """ 持仓跟踪智能体（AI Agent + 工作流） 对持仓/自选股票进行定期深度跟踪，工作流： 1. 上下文构建 读取产业链知识库（上游/下游/同业 + 关键词） 2. 数据采集 个股动态 + 上游产业链 + 下游产业链 + 同业动态（DB + RAG 向量检索） 3. 智能体分析 大模型扮演产业链跟踪分析师，输出深度专业分析 + 影响度判定 4. 沉淀与通知 结果入库 tracking_reports；重大变化（impact_score 达阈值）邮件通知 「智能体」体现在：大模型自主综合个股与产业链多环节信息，输出结构化的 个股动态 / 上游供给成本 / 下游需求景气 / 同业竞争 / 传导影响 / 风险与关注要点， 并给出 0-100 影响度评分与变化性质判定。 """ import json import logging import re import threading import time from database import query, query_one, execute from settings import tracking_config, tracking_state, set_tracking_state, mail_config, \ quiet_config, in_quiet_period from engine.chain_data import get_chain from engine.indicators import compute_indicators from rag.vector_store import query_vectors from config import CHROMA_NEWS_COLLECTION from engine.analyst import llm_chat log = logging.getLogger("agent") _track_lock = threading.Lock() _cycle_running = False _jobs = {} # code -> {status, error, ts} # ===================================================================== 数据采集 def _db_news(codes, days=40, limit=8): """按关联股票代码取新闻（近 days 天）""" if not codes: return [] conds, args = [], [] for c in codes: conds.append("(related_stocks=? OR related_stocks LIKE ? OR related_stocks LIKE ?)") args += [c, f"%,{c}", f"{c},%"] args.append(limit) return query( f"SELECT id,title,content,source,category,publish_date,sentiment FROM news " f"WHERE ({' OR '.join(conds)}) AND publish_date >= date('now','-{days} day') " f"ORDER BY publish_date DESC LIMIT ?", args) def _rag_news_text(question, where=None, top_k=4): """向量语义检索，返回紧凑文本""" try: hits = query_vectors(question, n_results=top_k, where=where, name=CHROMA_NEWS_COLLECTION) out = [] for h in hits: m = h.get("metadata", {}) out.append(f" [{m.get('date','')}] {m.get('title','')} (情感{m.get('sentiment',0):+.2f}) " f"{h.get('document','')[:90]}") return out except Exception as e: log.warning("rag fail: %s", e) return [] def _fmt_news(items): return "\n".join( f" [{n['publish_date']}] {n['title']} (情感{n['sentiment']:+.2f}) {n['content'][:90]}" for n in items) or " （暂无）" def collect_chain(code): """采集个股 + 产业链各环节资讯，返回结构化 dict""" stock = query_one("SELECT * FROM stocks WHERE code=?", (code,)) if not stock: return None chain = get_chain(stock["industry"]) seg = {} # 1. 个股直接动态 direct = _db_news([code], days=40, limit=8) direct_rag = _rag_news_text(f"{stock['name']} 最新动态 业绩 公告 重大事项", where={"code": code}, top_k=4) seg["direct"] = {"db": _fmt_news(direct), "rag": "\n".join(direct_rag) or " （暂无）"} # 2. 上游 up_db = _db_news(chain["upstream"]["codes"], days=40, limit=6) up_rag = _rag_news_text(f"{stock['industry']} 上游 {chain['upstream']['keywords']}", top_k=4) seg["upstream"] = {"codes": chain["upstream"]["codes"], "keywords": chain["upstream"]["keywords"], "db": _fmt_news(up_db), "rag": "\n".join(up_rag) or " （暂无）"} # 3. 下游 dn_db = _db_news(chain["downstream"]["codes"], days=40, limit=6) dn_rag = _rag_news_text(f"{stock['industry']} 下游需求 景气 {chain['downstream']['keywords']}", top_k=4) seg["downstream"] = {"codes": chain["downstream"]["codes"], "keywords": chain["downstream"]["keywords"], "db": _fmt_news(dn_db), "rag": "\n".join(dn_rag) or " （暂无）"} # 4. 同业 peer_db = _db_news(chain["peers"], days=40, limit=5) peer_rag = _rag_news_text(f"{stock['industry']} 竞争格局 同业 {chain['peers']}", top_k=3) seg["peers"] = {"codes": chain["peers"], "db": _fmt_news(peer_db), "rag": "\n".join(peer_rag) or " （暂无）"} # 5. 技术面 + 机构 ind = compute_indicators(query("SELECT date,open,high,low,close,volume FROM stock_daily " "WHERE code=? ORDER BY date ASC", (code,))) ratings = query("SELECT inst_name, rating, target_price, rating_date FROM inst_ratings " "WHERE stock_code=? ORDER BY rating_date DESC LIMIT 5", (code,)) holdings = query("SELECT inst_name, quarter, hold_value, change_pct FROM fund_holdings " "WHERE stock_code=? ORDER BY quarter DESC LIMIT 5", (code,)) seg["stock"] = stock seg["chain"] = chain seg["indicators"] = ind seg["ratings"] = ratings seg["holdings"] = holdings return seg # ===================================================================== 分析提示词 def _fmt_ind(ind): return (f"最新价 {ind.get('close')}（{ind.get('change_pct',0):+.2f}%），5日{ind.get('chg_5d',0):+.2f}% / " f"20日{ind.get('chg_20d',0):+.2f}%，RSI={ind.get('rsi')}，量比{ind.get('vol_ratio')}，" f"MA20={ind.get('ma20')}") def build_prompt(seg): s = seg["stock"] chain = seg["chain"] rated = "、".join(f"{r['inst_name']}({r['rating']},目标{r['target_price']})" for r in seg["ratings"]) or "暂无" held = "；".join(f"{h['inst_name']} {h['quarter']}持仓{h['hold_value']:.0f}万 环比{h['change_pct']:+.1f}%" for h in seg["holdings"]) or "暂无" return f"""你是资深产业链跟踪分析师，正在对【持仓标的】{s['name']}({s['code']}) 进行深度跟踪。请综合【个股】与【产业链上下游/同业】的全部动态，输出一份专业、有洞察的产业链跟踪分析。 【个股基本面】 {s.get('description','')} 【技术面】{_fmt_ind(seg['indicators'])} 【机构动向】评级：{rated} ｜ 持仓：{held} 【一、个股直接动态】（公告/新闻/机构观点） {seg['direct']['db']} {seg['direct']['rag']} 【二、上游产业链】（供给/原材料/成本端；关联股票 {seg['upstream']['codes'] or '无'}，关键词：{seg['upstream']['keywords']}） {seg['upstream']['db']} {seg['upstream']['rag']} 【三、下游产业链】（需求/客户/景气端；关联股票 {seg['downstream']['codes'] or '无'}，关键词：{seg['downstream']['keywords']}） {seg['downstream']['db']} {seg['downstream']['rag']} 【四、同业动态】（竞争格局；{seg['peers']['codes'] or '无'}） {seg['peers']['db']} {seg['peers']['rag']} 【输出要求】 第一步，先输出一个 json 代码块（必须最先输出，内容为本次判定，不要包含其他内容）： ```json {{"significance":"high|medium|low","impact_score":0到100的整数,"change_kind":"利好/利空/中性/震荡","summary":"一句话总结","chain_trend":"产业链趋势判断"}} ``` 第二步，再输出 Markdown 分析报告，结构如下： ## 一、个股最新动态 ## 二、上游产业链分析（供给、原材料、成本端变化及其传导） ## 三、下游产业链分析（需求、客户、景气度变化及其传导） ## 四、同行业竞争格局 ## 五、产业链传导与投资启示（上游→中游→下游，对{ s['name']}的影响路径） ## 六、风险提示 ## 关注要点（3-5条） 分析须严格基于提供的资讯，避免编造。impact_score 反映本次跟踪发现的动态对股价的潜在影响程度：>=65 视为重大变化。""" def parse_judge(text): """从容错地从 LLM 输出中提取 JSON 判定（支持 json 代码块/截断/缺失字段）""" t = text.strip() # 0) 优先取 json 代码块 m = re.search(r"```json\s*(.*?)\s*```", t, re.S) if m: try: j = json.loads(m.group(1)) if "significance" in j or "impact_score" in j: return j except Exception: pass # 1) 整体 JSON 对象解析 for m in re.finditer(r"\{[^{}]*\}", t, re.S): try: j = json.loads(m.group(0)) if "significance" in j or "impact_score" in j: return j except Exception: continue # 2) 逐字段容错提取（末尾被截断时也能拿到已输出字段） out = {} m = re.search(r'"significance"\s*:\s*"(high|medium|low)"', t) if m: out["significance"] = m.group(1) m = re.search(r'"impact_score"\s*:\s*(\d+)', t) if m: out["impact_score"] = int(m.group(1)) m = re.search(r'"change_kind"\s*:\s*"([^"]{1,20})"', t) if m: out["change_kind"] = m.group(1) m = re.search(r'"summary"\s*:\s*"((?:[^"\\]|\\.){1,200})"', t) if m: out["summary"] = m.group(1) m = re.search(r'"chain_trend"\s*:\s*"((?:[^"\\]|\\.){1,120})"', t) if m: out["chain_trend"] = m.group(1) return out or None def strip_json_block(text): """从报告文本中剥离最前面的 json 代码块，保留纯 Markdown""" m = re.search(r"```json\s*.*?```\s*", text, re.S) if m: return text[m.end():].strip() return text # ===================================================================== 执行 def track_stock(code, focus=""): """执行一次跟踪，返回 {ok, report_id, meta, ...}""" with _track_lock: seg = collect_chain(code) if not seg: return {"error": "股票不存在"} s = seg["stock"] prompt = build_prompt(seg) try: reply = llm_chat([ {"role": "system", "content": "你是一名严谨专业的产业链跟踪分析师，输出结构化、有数据支撑的分析。"}, {"role": "user", "content": prompt}, ]).strip() if not reply: raise RuntimeError("LLM 返回为空") judge = parse_judge(reply) # 摘要兜底：解析不到则取报告首个标题；并剥离 json 块保留纯净 Markdown clean_report = strip_json_block(reply) summary = (judge or {}).get("summary") or _first_heading(clean_report) meta = { "significance": (judge or {}).get("significance", "medium"), "impact_score": int((judge or {}).get("impact_score", 50)), "change_kind": (judge or {}).get("change_kind", "中性"), "summary": summary, "chain_trend": (judge or {}).get("chain_trend", ""), "news_counts": { "direct": _count(seg["direct"]), "upstream": _count(seg["upstream"]), "downstream": _count(seg["downstream"]), "peers": _count(seg["peers"]), }, "focus": focus, } sources = { "direct": seg["direct"], "upstream": seg["upstream"], "downstream": seg["downstream"], "peers": seg["peers"], "indicators": _fmt_ind(seg["indicators"]), "ratings": seg["ratings"], "holdings": seg["holdings"], } execute( "INSERT INTO tracking_reports(target_type, code, stock_name, industry, report, meta, sources, status, created_at) " "VALUES('stock',?,?,?,?,?,?,'done',datetime('now','localtime'))", (code, s["name"], s["industry"], clean_report, json.dumps(meta, ensure_ascii=False), json.dumps(sources, ensure_ascii=False))) rid = query_one("SELECT MAX(id) id FROM tracking_reports")["id"] _notify_if_significant(rid, s, meta) return {"ok": True, "report_id": rid, "meta": meta} except Exception as e: log.exception("track %s fail", code) return {"error": str(e)} def _count(seg): return (seg["db"].count("[") + seg["rag"].count("[")) // 1 def _first_heading(text): """取报告第一行非空文本作为摘要兜底""" for line in (text or "").splitlines(): line = line.strip().lstrip("#* ").strip() if line: return line[:60] return "" def _notify_if_significant(rid, stock, meta): """影响度达阈值且开启通知 → 邮件（股票/概念通用）。静默期内抑制发送。""" cfg = tracking_config() try: if cfg["notify"] and int(meta.get("impact_score", 0)) >= int(cfg["impact_threshold"]): qc = quiet_config("tracking") if qc["enabled"] and in_quiet_period(qc["ranges"]): log.info("tracking notify suppressed by quiet period: %s", stock.get("name")) return from engine.notifier import send_email mc = mail_config() nc = meta.get("news_counts") or {} detail = (f"个股资讯 {nc.get('direct', 0)} 条 / 上游 {nc.get('upstream', 0)} 条 / " f"下游 {nc.get('downstream', 0)} 条 / 同业 {nc.get('peers', 0)} 条" if "upstream" in nc else f"相关资讯 {nc.get('direct', 0)} 条 / 语义检索 {nc.get('rag', 0)} 条 / " f"受益个股 {nc.get('related', 0)} 只") send_email( f"[持仓跟踪] {stock['name']} 出现{meta.get('change_kind', '')}动态（影响度{meta.get('impact_score', 0)}）", f"""
🧭 智能体跟踪 · {stock['name']}（{stock['code']}）

影响度：{meta.get('impact_score', 0)}/100（{'🔴 重大' if int(meta.get('impact_score', 0)) >= 65 else '🟡 关注'}）
性质：{meta.get('change_kind', '')} ｜ 显著性：{meta.get('significance', '')}

摘要：{meta.get('summary', '')}

趋势判断：{meta.get('chain_trend', '')}

{detail}

""", cfg=mc) set_tracking_state(last_alert=int(meta.get("impact_score", 0))) except Exception as e: log.warning("track notify fail: %s", e) # ===================================================================== 概念/主题跟踪 def track_concept(name, keywords="", focus=""): """跟踪一个概念/主题：采集相关新闻 → 受益个股梳理 → LLM 深度分析 → 影响度判定""" kw = keywords.strip() or name kw_list = [k.strip() for k in kw.replace("，", ",").split(",") if k.strip()] kws = " ".join(kw_list) # 1) 采集：DB 关键词检索 + RAG 向量检索 conds, args = [], [] for k in kw_list[:5]: conds.append("(title LIKE ? OR content LIKE ?)") args += [f"%{k}%", f"%{k}%"] args.append(15) db_news = query( f"SELECT id,title,content,source,category,publish_date,sentiment,related_stocks FROM news " f"WHERE ({' OR '.join(conds)}) AND publish_date >= date('now','-45 day') " f"ORDER BY publish_date DESC LIMIT ?", args) rag_hits = _rag_news_text(f"{name} 概念主题 {kws} 最新动态 政策 催化", top_k=8) db_text = _fmt_news(db_news) rag_text = "\n".join(rag_hits) or " （暂无）" # 2) 受益个股：从 DB 新闻 related_stocks + RAG 命中元数据 code 汇总 codes = set() for n in db_news: for c in (n["related_stocks"] or "").split(","): if c: codes.add(c) try: for h in query_vectors(f"{name} {kws} 受益个股", n_results=8, name=CHROMA_NEWS_COLLECTION): c = (h.get("metadata") or {}).get("code") if c: codes.add(c) except Exception: pass related = query("SELECT code, name, industry FROM stocks WHERE code IN (%s)" % ",".join(["?"] * len(codes)), tuple(codes)) if codes else [] related_txt = "、".join(f"{r['name']}({r['code']},{r['industry']})" for r in related) or "暂无" # 3) 提示词 prompt = f"""你是资深题材/概念跟踪分析师，正在深度跟踪【{name}】这一概念主题。 【概念/主题】{name} 【检索关键词】{kws} 【最新相关资讯】 {db_text} {rag_text} 【关联受益个股】{related_txt} 【输出要求】 第一步，先输出一个 json 代码块（必须最先输出）： ```json {{"significance":"high|medium|low","impact_score":0到100的整数,"change_kind":"利好/利空/中性/震荡","summary":"一句话总结","chain_trend":"题材趋势判断"}} ``` 第二步，输出 Markdown 分析报告，结构如下： ## 一、概念主题最新动态 ## 二、核心驱动与催化（政策/产业/事件） ## 三、受益标的梳理（关联个股及逻辑） ## 四、市场情绪与资金动向 ## 五、风险提示 ## 关注要点（3-5条） 分析须严格基于资讯，避免编造。impact_score >=65 视为重大变化。""" try: reply = llm_chat([ {"role": "system", "content": "你是一名严谨专业的题材与概念跟踪分析师。"}, {"role": "user", "content": prompt}, ]).strip() if not reply: raise RuntimeError("LLM 返回为空") judge = parse_judge(reply) clean_report = strip_json_block(reply) summary = (judge or {}).get("summary") or _first_heading(clean_report) meta = { "significance": (judge or {}).get("significance", "medium"), "impact_score": int((judge or {}).get("impact_score", 50)), "change_kind": (judge or {}).get("change_kind", "中性"), "summary": summary, "chain_trend": (judge or {}).get("chain_trend", ""), "news_counts": {"direct": len(db_news), "rag": len(rag_hits), "related": len(related)}, "keywords": kws, } sources = {"keywords": kws, "db_news": db_text, "rag_news": rag_text, "related": related_txt} cid = f"CONCEPT:{name}" execute( "INSERT INTO tracking_reports(target_type, code, stock_name, industry, report, meta, sources, status, created_at) " "VALUES('concept',?,?,?,?,?,?,'done',datetime('now','localtime'))", (cid, name, "概念/主题", clean_report, json.dumps(meta, ensure_ascii=False), json.dumps(sources, ensure_ascii=False))) rid = query_one("SELECT MAX(id) id FROM tracking_reports")["id"] _notify_if_significant(rid, {"name": name, "code": cid}, meta) return {"ok": True, "report_id": rid, "meta": meta} except Exception as e: log.exception("track concept %s fail", name) return {"error": str(e)} # ===================================================================== 目标管理 def list_targets(): return query("SELECT * FROM watch_targets ORDER BY type, id") def add_target(ttype, name, code="", keywords=""): name = name.strip() if not name: return {"error": "名称不能为空"} execute("INSERT INTO watch_targets(type, code, name, keywords) VALUES(?,?,?,?)", (ttype, code, name, keywords)) return {"ok": True} def delete_target(tid): execute("DELETE FROM watch_targets WHERE id=?", (tid,)) return {"ok": True} # ===================================================================== 批量与调度 def track_all(progress=None): """跟踪全部目标：持仓（自选股）+ 目标(概念/主题/股票)""" global _cycle_running if _cycle_running: return {"tracked": 0, "msg": "已有跟踪任务进行中，请稍后再试"} _cycle_running = True try: targets = [] # 持仓（自选股） for w in query("SELECT w.code, s.name FROM watchlist w JOIN stocks s ON s.code=w.code ORDER BY w.added_at"): targets.append(("stock", w["code"], w["name"])) # 目标（股票/概念） for t in query("SELECT id, type, code, name, keywords FROM watch_targets WHERE enabled=1 ORDER BY id"): if t["type"] == "stock" and t["code"]: targets.append(("stock", t["code"], t["name"])) else: targets.append(("concept", t["name"], t["keywords"] or t["name"])) # 去重 seen, uniq = set(), [] for typ, key, name in targets: u = (typ, key) if u in seen: continue seen.add(u) uniq.append((typ, key, name)) if not uniq: return {"tracked": 0, "msg": "暂无跟踪目标：请添加持仓/自选股或概念主题目标"} results = [] for i, (typ, key, name) in enumerate(uniq): if typ == "stock": r = track_stock(key) else: r = track_concept(key, name) results.append({"type": typ, "name": name, **r}) set_tracking_state(last_run=time.strftime("%Y-%m-%d %H:%M:%S"), last_stock=name) if progress: progress(i + 1, len(uniq)) return {"tracked": len(results), "results": results} finally: _cycle_running = False def latest_reports(code, limit=5): return query("SELECT id, target_type, code, stock_name, industry, meta, status, created_at " "FROM tracking_reports WHERE code=? ORDER BY id DESC LIMIT ?", (code, limit)) def list_reports(limit=30): return query("SELECT id, target_type, code, stock_name, industry, meta, status, created_at " "FROM tracking_reports ORDER BY id DESC LIMIT ?", (limit,)) def get_report(rid): return query_one("SELECT * FROM tracking_reports WHERE id=?", (rid,)) class TrackingThread(threading.Thread): """后台调度：定期跟踪全部持仓股票""" def __init__(self): super().__init__(daemon=True, name="tracking") self._stop = threading.Event() def stop(self): self._stop.set() def run(self): log.info("持仓跟踪调度器启动") first = True while not self._stop.is_set(): try: cfg = tracking_config() if first: # 启动后等待一个完整间隔再首跑，避免重启即烧一轮 LLM、与手动操作冲突 self._stop.wait(cfg.get("interval_min", 60) * 60) first = False continue if cfg["enabled"]: try: r = track_all() log.info("tracking cycle: %s", r) except Exception as e: log.warning("tracking cycle error: %s", e) except Exception as e: log.warning("tracking loop error: %s", e) self._stop.wait(cfg.get("interval_min", 60) * 60) log.info("持仓跟踪调度器停止") _tracking = None def start_tracking(): global _tracking if _tracking and _tracking.is_alive(): return _tracking _tracking = TrackingThread() _tracking.start() return _tracking