金融机构如何让 4000 人用上可信 AI——LSEG 的规模化实践

"""
模拟金融洞察发布流水线：数据校验 → AI 生成 → 输出审计 → 人工确认
运行前需设置环境变量：export OPENAI_API_KEY="sk-..."
依赖：pip install openai pydantic
"""

import os
import json
from datetime import datetime
from openai import OpenAI
from pydantic import BaseModel, Field, validator

# ---- 1. 数据输入模型：带来源标记和校验 ----
class MarketDataInput(BaseModel):
    symbol: str = Field(..., description="证券代码，如 .LON 后缀")
    price: float = Field(..., gt=0, description="当前价格，必须为正数")
    volume: int = Field(..., ge=0, description="成交量")
    source: str = Field(..., description="数据来源标识，如 'LSEG-RealTime'")
    timestamp: datetime

    @validator("symbol")
    def symbol_must_have_exchange(cls, v):
        if "." not in v:
            raise ValueError("证券代码必须包含交易所后缀，如 HSBA.LON")
        return v

# ---- 2. AI 洞察生成：带系统提示约束输出格式 ----
client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])

TRUSTED_SYSTEM_PROMPT = """你是 LSEG 金融分析助手。规则：
1. 只基于用户提供的数据进行分析，不得编造数据。
2. 每条洞察必须标注置信度（high/medium/low）和依据的数据字段。
3. 如果数据不足以得出结论，明确说明而非推测。
4. 输出 JSON 格式，包含 insight、confidence、data_basis 三个字段。"""

def generate_insight(data: MarketDataInput) -> dict:
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": TRUSTED_SYSTEM_PROMPT},
            {"role": "user", "content": f"分析以下市场数据：{data.json()}"}
        ],
        response_format={"type": "json_object"},
        temperature=0.3  # 低温度减少创造性偏差
    )
    result = json.loads(response.choices[0].message.content)
    # 强制附加溯源元数据
    result["source_tag"] = data.source
    result["generated_at"] = datetime.now().isoformat()
    result["model_version"] = "gpt-4o-2024-08-06"
    return result

# ---- 3. 输出审计：自动检查合规红线 ----
def audit_insight(insight: dict) -> tuple[bool, str]:
    """检查 AI 输出是否触碰信任红线"""
    blockers = []
    # 红线 1：不能给出具体买卖建议（合规要求）
    buy_keywords = ["建议买入", "应该卖出", "强烈推荐", "must buy", "should sell"]
    text = insight.get("insight", "").lower()
    if any(k in text for k in buy_keywords):
        blockers.append("包含具体买卖建议，违反合规要求")
    # 红线 2：置信度为 low 时必须标记需人工复核
    if insight.get("confidence") == "low":
        blockers.append("置信度低，需人工复核")
    # 红线 3：数据依据字段不能为空
    if not insight.get("data_basis"):
        blockers.append("缺少数据依据说明")
    return (len(blockers) == 0, "; ".join(blockers) if blockers else "通过")

# ---- 4. 运行完整流水线 ----
sample_data = MarketDataInput(
    symbol="HSBA.LON",
    price=7.42,
    volume=12500000,
    source="LSEG-RealTime-Equities",
    timestamp=datetime.now()
)

insight = generate_insight(sample_data)
passed, reason = audit_insight(insight)

print("=" * 50)
print("【AI 洞察输出】")
print(json.dumps(insight, indent=2, ensure_ascii=False))
print(f"\n【审计结果】{'✅ ' + reason if passed else '❌ ' + reason}")
if not passed:
    print("⚠️  该洞察已被自动拦截，需人工复核后方可发布。")

# insight_roles.yaml — 角色化洞察分发配置
roles:
  analyst:
    allowed_data_sources: ["LSEG-RealTime-Equities", "LSEG-Fundamentals"]
    output_format: "structured_summary"  # 结构化摘要，含置信度
    max_insights_per_hour: 50
    audit_level: "standard"              # 标准审计，记录输入输出

  trader:
    allowed_data_sources: ["LSEG-RealTime-Equities"]
    output_format: "signal_card"         # 信号卡片，只显示关键指标和方向
    max_insights_per_hour: 200
    audit_level: "enhanced"              # 增强审计，额外记录决策上下文

  compliance_officer:
    allowed_data_sources: ["*"]          # 全量数据源访问
    output_format: "audit_log"           # 审计日志格式
    max_insights_per_hour: unlimited
    audit_level: "full_trace"            # 全链路追溯，含模型版本和提示词

trust_gates:
  - name: "no_buy_sell_advice"
    action: "block"
    description: "禁止具体买卖建议"

  - name: "low_confidence_flag"
    action: "flag_for_review"
    description: "低置信度标记需人工复核"

  - name: "data_basis_required"
    action: "block"
    description: "必须包含数据依据说明"

model_config:
  model: "gpt-4o"
  temperature: 0.3
  max_tokens: 1024
  response_format: "json_object"