145 lines
4.3 KiB
Python
145 lines
4.3 KiB
Python
"""LLM 翻译 + 投资事件抽取数据模型 (M4)。
|
|
|
|
EnTranslatedArticle 是 M4 最终落盘格式，包含中英文双语内容和抽取的事件。
|
|
EventExtraction 是 LLM JSON 输出直接映射，经 Pydantic 强校验。
|
|
"""
|
|
|
|
import re
|
|
from datetime import datetime
|
|
from enum import StrEnum
|
|
|
|
from pydantic import BaseModel, Field, field_validator
|
|
|
|
|
|
class Sentiment(StrEnum):
|
|
"""事件情绪倾向。"""
|
|
|
|
POSITIVE = "positive" # 利好
|
|
NEUTRAL = "neutral" # 中性
|
|
NEGATIVE = "negative" # 利空
|
|
|
|
|
|
# 国际财经事件类型（LLM Prompt 中展示）
|
|
INTERNATIONAL_EVENT_TYPES: tuple[str, ...] = (
|
|
"财报披露",
|
|
"并购收购",
|
|
"产品发布",
|
|
"监管政策",
|
|
"宏观经济",
|
|
"央行决议",
|
|
"行业动态",
|
|
"技术突破",
|
|
"高管变动",
|
|
"诉讼法律",
|
|
"市场异动",
|
|
"地缘政治",
|
|
"大宗商品",
|
|
"外汇波动",
|
|
"其他",
|
|
)
|
|
|
|
# 美股代码正则：1-5 个大写字母
|
|
_US_STOCK_RE = re.compile(r"^[A-Z]{1,5}$")
|
|
|
|
MIN_IMPORTANCE = 1
|
|
MAX_IMPORTANCE = 5
|
|
|
|
|
|
class EventExtraction(BaseModel):
|
|
"""LLM 输出的单个事件，直接映射 JSON。"""
|
|
|
|
event_type: str = Field(..., description="事件类型，见 INTERNATIONAL_EVENT_TYPES")
|
|
stock_codes: list[str] = Field(
|
|
default_factory=list,
|
|
description="涉及美股代码，如 AAPL、TSLA；无相关股票时为空",
|
|
)
|
|
sentiment: Sentiment = Field(..., description="positive/neutral/negative")
|
|
importance: int = Field(
|
|
..., ge=MIN_IMPORTANCE, le=MAX_IMPORTANCE, description="1-5 重要程度"
|
|
)
|
|
summary_zh: str = Field(
|
|
default="",
|
|
max_length=200,
|
|
description="一句话中文事件摘要",
|
|
)
|
|
|
|
@field_validator("stock_codes")
|
|
@classmethod
|
|
def _validate_stock_codes(cls, v: list[str]) -> list[str]:
|
|
"""剔除非美股代码格式、去重、统一大写。"""
|
|
cleaned: list[str] = []
|
|
seen: set[str] = set()
|
|
for code in v:
|
|
s = (code or "").strip().upper()
|
|
if not s or not _US_STOCK_RE.match(s):
|
|
continue
|
|
if s not in seen:
|
|
seen.add(s)
|
|
cleaned.append(s)
|
|
return cleaned
|
|
|
|
@field_validator("event_type")
|
|
@classmethod
|
|
def _normalize_event_type(cls, v: str) -> str:
|
|
s = (v or "").strip()
|
|
return s if s else "其他"
|
|
|
|
|
|
class LLMTranslationOutput(BaseModel):
|
|
"""LLM 单次调用的完整输出 JSON 映射。"""
|
|
|
|
title_zh: str = Field(..., description="中文翻译标题")
|
|
content_zh: str = Field(..., description="中文翻译正文")
|
|
events: list[EventExtraction] = Field(
|
|
default_factory=list, description="提取的投资事件列表"
|
|
)
|
|
|
|
|
|
class EnTranslatedArticle(BaseModel):
|
|
"""M4 最终落盘格式：双语文章 + 抽取事件 + 调用元信息。"""
|
|
|
|
# ── 来源标识 ──
|
|
source_id: str
|
|
source_name: str
|
|
url: str
|
|
url_hash: str
|
|
|
|
# ── 双语内容 ──
|
|
title: str = "" # 英文原标题
|
|
title_zh: str = "" # 中文翻译标题
|
|
content_en: str = "" # 英文原文
|
|
content_zh: str = "" # 中文翻译
|
|
publish_time: str = "" # ISO 8601
|
|
word_count: int = 0 # 英文词数
|
|
word_count_zh: int = 0 # 中文译文字数
|
|
|
|
# ── 抽取事件 ──
|
|
events: list[EventExtraction] = Field(default_factory=list)
|
|
|
|
# ── 调用元信息 ──
|
|
provider: str = "" # deepseek / qwen
|
|
model: str = ""
|
|
translated_at: datetime = Field(default_factory=datetime.now)
|
|
attempts: int = 1 # 实际调用次数（含重试）
|
|
prompt_tokens: int | None = None
|
|
completion_tokens: int | None = None
|
|
|
|
def short_summary(self) -> str:
|
|
codes = set()
|
|
for ev in self.events:
|
|
codes.update(ev.stock_codes)
|
|
codes_str = ",".join(sorted(codes)[:5]) or "-"
|
|
return (
|
|
f"[{self.source_id}] {self.title[:30]}... "
|
|
f"→ {len(self.events)}events, stocks: {codes_str}"
|
|
)
|
|
|
|
|
|
class LLMCallError(Exception):
|
|
"""LLM 调用失败（网络 / 解析 / 校验）。"""
|
|
|
|
def __init__(self, reason: str, *, attempts: int = 0) -> None:
|
|
super().__init__(reason)
|
|
self.reason = reason
|
|
self.attempts = attempts
|