38 lines
1.3 KiB
Python
38 lines
1.3 KiB
Python
"""轻量 RAG：检索与个股/大盘相关的资讯，做情绪标注，作为 LLM 上下文降低幻觉。
|
||
|
||
当前为基于来源接口 + 关键词情绪的检索式上下文；后续可平滑升级为向量检索（embedding + 向量库）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import akshare_service as svc
|
||
|
||
|
||
def stock_news(symbol: str, limit: int = 5):
|
||
"""返回个股相关资讯（已带利好/利空标注）。"""
|
||
try:
|
||
data = svc.get_stock_news(symbol, limit=limit)
|
||
return data.get("list", [])[:limit]
|
||
except Exception:
|
||
return []
|
||
|
||
|
||
def _senti_score(items):
|
||
pos = sum(1 for x in items if x.get("sentiment") == "利好")
|
||
neg = sum(1 for x in items if x.get("sentiment") == "利空")
|
||
if pos > neg:
|
||
return "利好", pos, neg
|
||
if neg > pos:
|
||
return "利空", pos, neg
|
||
return "中性", pos, neg
|
||
|
||
|
||
def stock_context(symbol: str, limit: int = 5):
|
||
"""供 AI 诊断使用：检索资讯 + 汇总情绪。"""
|
||
items = stock_news(symbol, limit)
|
||
tone, pos, neg = _senti_score(items)
|
||
block = ""
|
||
if items:
|
||
block = "近期相关资讯（检索）：\n" + "\n".join(
|
||
f"- [{x.get('sentiment','中性')}] {x.get('title','')}（{x.get('time','')}）" for x in items)
|
||
return {"items": items, "tone": tone, "pos": pos, "neg": neg, "block": block}
|