【AI+量化实战 #04】研报预测能信吗？让AI当你的业绩预告审计员

import mairui
import os

api = mairui.Client("<你的证书>")  # 演示证书，请替换成你自己的正式证书
# 业绩预告
forecasts = api.company_earnings_forecast("600519")
for f in forecasts[:3]:
    print(f["rdate"], f["type"], f["abs"][:60])

# 实际利润
income = api.stock_income("600519", lt=4)
for i in income:
    print(i["jzrq"], round(float(i["jlr"]) / 1e8, 2), "亿元")

import re

_PROFIT_RE = re.compile(
    r"净利润为[：:]\s*([-]?\d+(?:\.\d+)?)\s*(亿元|万元)"
    r"(?:至([-]?\d+(?:\.\d+)?)\s*(亿元|万元))?"
)
_GROWTH_RE = re.compile(
    r"(?:变动幅度|增长|增幅)[：:]\s*([-]?\d+(?:\.\d+)?)\s*%"
    r"(?:至([-]?\d+(?:\.\d+)?)\s*%)?"
)

def parse_forecast(abs_text: str) -> dict:
    result = {"profit_low": None, "profit_high": None,
              "growth_low": None, "growth_high": None}
    if not abs_text:
        return result
    m = _PROFIT_RE.search(abs_text)
    if m:
        low_val = float(m.group(1))
        low_unit = m.group(2)
        result["profit_low"] = low_val * (10000 if low_unit == "亿元" else 1)
        if m.group(3):
            high_val = float(m.group(3))
            high_unit = m.group(4)
            result["profit_high"] = high_val * (10000 if high_unit == "亿元" else 1)
        else:
            result["profit_high"] = result["profit_low"]
    m = _GROWTH_RE.search(abs_text)
    if m:
        result["growth_low"] = float(m.group(1))
        result["growth_high"] = float(m.group(2)) if m.group(2) else result["growth_low"]
    return result

def audit_forecasts(forecast_df, actual_df):
    audits = []
    for _, f in forecast_df.iterrows():
        if not f["profit_low_wan"]:
            continue
        # 找同期实际利润
        rdate = f["rdate"]
        actual = actual_df[
            (actual_df["code"] == f["code"]) &
            (actual_df["report_date"].str.startswith(rdate[:4] if rdate else ""))
        ]
        issues = []
        if len(actual) > 0:
            actual_profit = actual.iloc[0]["net_profit_wan"]
            f_low = f["profit_low_wan"]
            f_high = f["profit_high_wan"] or f_low
            # 检查实际利润是否在预告区间内
            if actual_profit < f_low * 0.9 or actual_profit > f_high * 1.1:
                issues.append(
                    f"实际利润 {actual_profit:.0f}万 偏离预告区间 "
                    f"[{f_low:.0f}, {f_high:.0f}]万 超过10%")
            else:
                issues.append(
                    f"实际利润 {actual_profit:.0f}万 在预告区间内")
        else:
            issues.append("无同期实际数据，无法校验")

        # 检查增速方向与预告类型是否矛盾
        if f["growth_low_pct"] is not None:
            g = f["growth_low_pct"]
            if f["type"] == "预亏" and g > 0:
                issues.append(f"预告类型'预亏'但增速为正({g}%)，存在矛盾")
            elif f["type"] == "预增" and g < 0:
                issues.append(f"预告类型'预增'但增速为负({g}%)，存在矛盾")

        audits.append({
            "code": f["code"], "name": f["name"],
            "rdate": rdate, "type": f["type"],
            "audit_result": "; ".join(issues),
        })
    return audits

def build_llm_prompt(stock_name, forecasts, actuals):
    prompt = f"""你是一位严谨的金融分析师审计员。请审计以下 {stock_name} 的业绩预告一致性。

## 业绩预告数据
"""
    for _, f in forecasts.iterrows():
        prompt += f"\n- {f['rdate']} | 类型:{f['type']} | 预告文本:{f['abs_text']}"

    prompt += "\n\n## 实际财报数据\n"
    for _, a in actuals.iterrows():
        prompt += f"\n- {a['report_date']} | 净利润:{a['net_profit_wan']:.0f}万 | 营收:{a['revenue_wan']:.0f}万"

    prompt += """

## 审计任务
1. 预告增速与实际增速是否一致？偏差超过 10% 请标红。
2. 预告利润区间是否包含实际利润？不在区间内请标红。
3. 预告类型（预增/预升/预亏）与数字方向是否矛盾？
4. 是否存在隐含假设（如"上年同期"基数是否被调整过）？
5. 跨期预告是否连贯（如 Q1 预告 vs 半年报预告的增速是否递进）？

## 输出格式
JSON: [{"issue": "...", "severity": "high/medium/low", "explanation": "..."}]
"""
    return prompt