# -*- coding: utf-8 -*- """ 定期报告分析解读引擎 v7.1.0 (report_interpreter.py) 把上市公司定期报告（年报/半年报/季报）的财务数据转化为**自然语言解读**： 业绩概览、同比/环比、盈利能力、成长性、现金流质量、亮点信号、风险警示、 综合评分与评级建议。纯规则引擎（阈值+模板），零 LLM 依赖、零 pip 依赖。 数据来源（多路输入）： 1. interpret_stock("600519") —— 东财 RPT_LICO_FN_CPD 直连（含同比/环比增速字段） 2. interpret_data([{...}, ...]) —— 直接喂财务数据 dict（离线/自定义，字段名兼容中英文） 3. interpret_periods(periods) —— 已归一化的周期序列 用法： from scripts.report_interpreter import interpret_stock_report text = interpret_stock_report("600519") # 一站式解读 result = interpret_data({"营收": 100, "净利润": 20, ...}) # 结构化结果 """ from __future__ import annotations import json import re import sys from datetime import datetime from typing import Any, Dict, List, Optional, Tuple try: from .http_utils import http_get except ImportError: # 脚本直跑 from http_utils import http_get # type: ignore # ── 东财财务报告接口（RPT_LICO_FN_CPD，含同比/环比增速字段）──────────── _EM_FN_URL = ("https://datacenter-web.eastmoney.com/api/data/v1/get" "?reportName=RPT_LICO_FN_CPD&columns=ALL" "&filter=(SECUCODE%3D%22{secucode}%22)" "&pageNumber=1&pageSize={limit}&source=WEB&client=WEB") _EM_HEADERS = { "User-Agent": ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/133.0.0.0 Safari/537.36"), "Referer": "https://data.eastmoney.com/", "Accept": "application/json, text/plain, */*", } # ── 解读规则（阈值 → 档位标签 + 评语模板）───────────────────────── RULES: Dict[str, Dict[str, Any]] = { "revenue_yoy": { # 营收同比增速（%） ">30": ("高增长", "营收同比大增 {v:.1f}%，规模扩张动能强劲"), "10~30": ("稳健增长", "营收同比增长 {v:.1f}%，处于稳健增长区间"), "0~10": ("低增长", "营收同比增长 {v:.1f}%，增速平缓"), "<0": ("营收下滑", "营收同比下滑 {v:.1f}%，需关注需求端压力"), }, "profit_yoy": { # 归母净利润同比增速（%） ">50": ("利润爆发", "归母净利润同比大增 {v:.1f}%，盈利爆发式增长"), "20~50": ("利润强劲", "归母净利润同比增长 {v:.1f}%，盈利强劲"), "0~20": ("利润平稳", "归母净利润同比增长 {v:.1f}%，盈利平稳"), "<0": ("利润承压", "归母净利润同比下滑 {v:.1f}%，盈利承压"), }, "gross_margin": { # 销售毛利率（%） ">60": ("极高毛利", "毛利率 {v:.1f}%，具备极强的定价权与成本优势"), "40~60": ("高毛利", "毛利率 {v:.1f}%，盈利空间充裕"), "25~40": ("中等毛利", "毛利率 {v:.1f}%，处于行业中游水平"), "<25": ("毛利偏低", "毛利率 {v:.1f}%，产品或面临同质化竞争"), }, "roe": { # 加权净资产收益率（%） ">20": ("ROE优秀", "ROE {v:.1f}%，资本回报水平优秀"), "12~20": ("ROE良好", "ROE {v:.1f}%，资本回报良好"), "6~12": ("ROE一般", "ROE {v:.1f}%，资本回报一般"), "<6": ("ROE偏弱", "ROE {v:.1f}%，资本回报偏弱"), }, "cash_ratio": { # 每股经营现金流 / 每股收益 ">1": ("现金流充沛", "每股经营现金流为 EPS 的 {v:.2f} 倍，盈利含金量高"), "0.5~1": ("现金流尚可", "每股经营现金流为 EPS 的 {v:.2f} 倍，盈利质量尚可"), "<0.5": ("现金流偏紧", "每股经营现金流仅 EPS 的 {v:.2f} 倍，盈利含金量不足"), }, } SCORE_WEIGHTS = { # 综合评分权重（合计 100） "profit_yoy": 25, "revenue_yoy": 20, "roe": 20, "gross_margin": 15, "cash_ratio": 10, "eps": 10, } def _band(value: float, rule: Dict[str, Any]) -> Tuple[str, str]: """按阈值区间取档位标签与评语模板。""" for key, (label, tpl) in rule.items(): ok = False if key == ">30": ok = value > 30 elif key == ">60": ok = value > 60 elif key == ">50": ok = value > 50 elif key == ">20": ok = value > 20 elif key == ">12": ok = value > 12 elif key == ">1": ok = value > 1 elif key == "10~30": ok = 10 <= value <= 30 elif key == "40~60": ok = 40 <= value <= 60 elif key == "25~40": ok = 25 <= value < 40 elif key == "12~20": ok = 12 <= value < 20 elif key == "6~12": ok = 6 <= value < 12 elif key == "0~10": ok = 0 <= value < 10 elif key == "0~20": ok = 0 <= value < 20 elif key == "0.5~1": ok = 0.5 <= value <= 1 elif key == "<0": ok = value < 0 elif key == "<25": ok = value < 25 elif key == "<6": ok = value < 6 elif key == "<0.5": ok = value < 0.5 if ok: return label, tpl return "数据缺失", "该指标本期数据缺失" def _f(v: Any) -> Optional[float]: """安全转 float（兼容 None/'-'/空串）。""" if v is None: return None if isinstance(v, (int, float)): return float(v) s = str(v).replace(",", "").replace("%", "").strip() if not s or s in ("-", "--", "nan", "None"): return None try: return float(s) except ValueError: return None # ── 字段归一化：东财字段名 / 中英文常见别名 → 统一键 ──────────────── _FIELD_ALIASES = { "revenue": ["TOTAL_OPERATE_INCOME", "OPERATE_INCOME", "营收", "营业收入", "营业总收入", "revenue", "total_revenue"], "net_profit": ["PARENT_NETPROFIT", "NETPROFIT", "归母净利润", "净利润", "net_profit", "netprofit"], "eps": ["BASIC_EPS", "EPSJB", "EPS", "每股收益", "基本每股收益"], "roe": ["WEIGHTAVG_ROE", "ROEJQ", "ROE", "净资产收益率", "加权净资产收益率"], "gross_margin": ["XSMLL", "毛利率", "销售毛利率", "gross_margin"], "revenue_yoy": ["YSTZ", "营收同比", "营业收入同比增长率", "TOTAL_OPERATE_INCOME_YOY"], "profit_yoy": ["SJLTZ", "净利同比", "净利润同比增长率", "PARENT_NETPROFIT_YOY"], "revenue_qoq": ["YSHZ", "营收环比", "营业收入环比增长率"], "profit_qoq": ["SJLHZ", "净利环比", "净利润环比增长率"], "cash_per_share": ["MGJYXJJE", "每股经营现金流", "每股经营现金流量净额"], "bps": ["BPS", "每股净资产"], "report_date": ["REPORTDATE", "REPORT_DATE", "报告期", "date"], "name": ["SECURITY_NAME_ABBR", "公司名称", "名称", "name"], } def normalize_period(row: Dict[str, Any]) -> Dict[str, Any]: """把任意来源的财务数据行归一化为统一键。""" out: Dict[str, Any] = {} for key, aliases in _FIELD_ALIASES.items(): for a in aliases: if a in row and row[a] is not None: v = row[a] if key in ("revenue", "net_profit", "eps", "roe", "gross_margin", "revenue_yoy", "profit_yoy", "revenue_qoq", "profit_qoq", "cash_per_share", "bps"): v = _f(v) out[key] = v break return out # ── 东财直连拉取 ────────────────────────────────────────────── def fetch_periods(stock_code: str, limit: int = 8) -> List[Dict[str, Any]]: """东财 RPT_LICO_FN_CPD 拉取财务周期（自动补 .SH/.SZ/.BJ）。""" code = stock_code.strip().upper() if "." in code: secucode, suffix = code.split(".")[:2] secucode = secucode + "." + suffix else: market = "BJ" if code.startswith(("4", "8", "92")) else ( "SH" if code.startswith(("6", "9")) else "SZ") secucode = f"{code}.{market}" try: resp = http_get(_EM_FN_URL.format(secucode=secucode, limit=limit), headers=_EM_HEADERS, timeout=20, rate_limit_delay=0.3) if resp is None: return [] data = (resp.json().get("result") or {}).get("data") or [] return [normalize_period(r) for r in data] except Exception as e: print(f"[report_interpreter] 拉取 {stock_code} 财务数据失败: {e}", file=sys.stderr) return [] # ── 解读核心 ────────────────────────────────────────────────── class ReportInterpreter: """定期报告解读器：指标计算 → 规则命中 → 评分评级 → 文本生成。""" def __init__(self, rules: Optional[Dict] = None, weights: Optional[Dict[str, int]] = None): self.rules = rules or RULES self.weights = weights or SCORE_WEIGHTS # ── 指标提取 ──────────────────────────────────────── @staticmethod def _latest(periods: List[Dict]) -> Dict[str, Any]: periods = [p for p in periods if p.get("report_date") or p.get("revenue")] if not periods: return {} # 报告期日期倒序取最新（无日期则取列表第一个） def _key(p: Dict) -> str: d = str(p.get("report_date") or "") return d.replace("-", "").replace("/", "").replace(" ", "") return sorted(periods, key=_key, reverse=True)[0] if any( _key(p) for p in periods) else periods[0] def compute(self, periods: List[Dict]) -> Dict[str, Any]: """计算指标与信号。""" latest = self._latest(periods) if not latest: return {"error": "无财务数据"} rev, np_ = latest.get("revenue"), latest.get("net_profit") eps, roe = latest.get("eps"), latest.get("roe") gm, cps = latest.get("gross_margin"), latest.get("cash_per_share") rev_yoy = latest.get("revenue_yoy") np_yoy = latest.get("profit_yoy") rev_qoq = latest.get("revenue_qoq") np_qoq = latest.get("profit_qoq") cash_ratio = None if cps is not None and eps is not None and eps > 0: cash_ratio = cps / eps # 负 EPS 时比率无意义（亏损公司直接看现金流正负） # 历史趋势：最近两期同比增速 trend = None if len(periods) >= 2: prev = periods[1] if (np_yoy is not None and prev.get("profit_yoy") is not None and np_yoy < 0 and prev.get("profit_yoy", 0) < 0): trend = "连续两期利润下滑" metrics = { "name": latest.get("name") or "", "report_date": latest.get("report_date") or "", "revenue": rev, "net_profit": np_, "eps": eps, "roe": roe, "gross_margin": gm, "revenue_yoy": rev_yoy, "profit_yoy": np_yoy, "revenue_qoq": rev_qoq, "profit_qoq": np_qoq, "cash_per_share": cps, "cash_ratio": cash_ratio, "bps": latest.get("bps"), "trend": trend, } return metrics # ── 规则命中 → 亮点/风险 ───────────────────────────── def signals(self, m: Dict[str, Any]) -> Tuple[List[str], List[str]]: highlights, risks = [], [] if m.get("error"): return highlights, risks # 营收/利润增速 if m.get("revenue_yoy") is not None: label, tpl = _band(m["revenue_yoy"], self.rules["revenue_yoy"]) text = tpl.format(v=m["revenue_yoy"]) (risks if m["revenue_yoy"] < 0 else highlights).append(text) if m.get("profit_yoy") is not None: label, tpl = _band(m["profit_yoy"], self.rules["profit_yoy"]) text = tpl.format(v=m["profit_yoy"]) (risks if m["profit_yoy"] < 0 else highlights).append(text) # 环比短期承压 for key, name in (("revenue_qoq", "营收"), ("profit_qoq", "净利润")): v = m.get(key) if v is not None and v < 0: risks.append(f"{name}环比下滑 {v:.1f}%，短期动能转弱") # 毛利率 if m.get("gross_margin") is not None: label, tpl = _band(m["gross_margin"], self.rules["gross_margin"]) text = tpl.format(v=m["gross_margin"]) (risks if m["gross_margin"] < 25 else highlights).append(text) # ROE if m.get("roe") is not None: label, tpl = _band(m["roe"], self.rules["roe"]) (risks if m["roe"] < 6 else highlights).append( tpl.format(v=m["roe"])) # 现金流质量 if m.get("cash_ratio") is not None: label, tpl = _band(m["cash_ratio"], self.rules["cash_ratio"]) text = tpl.format(v=m["cash_ratio"]) (risks if m["cash_ratio"] < 0.5 else highlights).append(text) elif m.get("cash_per_share") is not None and m.get("cash_per_share", 0) < 0: risks.append("每股经营现金流为负，经营性现金流出") # 趋势 if m.get("trend"): risks.append(m["trend"]) return highlights[:6], risks[:6] # ── 综合评分与评级 ─────────────────────────────────── def score(self, m: Dict[str, Any]) -> Tuple[float, str]: if m.get("error"): return 0.0, "无数据" total = 0.0 got = 0 # 各维度得分（0-100 按档位） def _dim(value, bands): for lo, hi, s in bands: if lo is None or value > lo: if hi is None or value <= hi: return s return 50 dims = { "profit_yoy": (m.get("profit_yoy"), [ (50, None, 95), (20, 50, 85), (0, 20, 65), (-10, 0, 40), (None, -10, 25)]), "revenue_yoy": (m.get("revenue_yoy"), [ (30, None, 90), (10, 30, 75), (0, 10, 55), (-10, 0, 35), (None, -10, 20)]), "roe": (m.get("roe"), [ (20, None, 90), (12, 20, 75), (6, 12, 55), (None, 6, 30)]), "gross_margin": (m.get("gross_margin"), [ (60, None, 90), (40, 60, 75), (25, 40, 55), (None, 25, 30)]), "cash_ratio": (m.get("cash_ratio"), [ (1, None, 90), (0.5, 1, 65), (None, 0.5, 30)]), "eps": (m.get("eps"), [ (1, None, 80), (0.3, 1, 65), (0, 0.3, 50), (None, 0, 25)]), } for key, weight in self.weights.items(): if key not in dims: continue value, bands = dims[key] if value is None: continue total += weight * _dim(value, bands) got += weight if got == 0: return 0.0, "数据不足" final = total / got if final >= 80: rating = "积极" elif final >= 65: rating = "中性偏积极" elif final >= 45: rating = "中性" else: rating = "谨慎" return round(final, 1), rating # ── 文本解读 ──────────────────────────────────────── def format_text(self, m: Dict[str, Any], highlights: List[str], risks: List[str], score: float, rating: str) -> str: if m.get("error"): return f"❌ 解读失败：{m['error']}" L = [] name = m.get("name") or "该公司" L.append(f"📊 定期报告解读：{name}" f"（报告期 {m.get('report_date') or '—'}）") L.append("=" * 52) if m.get("revenue") is not None: rev_txt = f"营收 {m['revenue']/1e8:.2f} 亿元" if m.get("revenue_yoy") is not None: rev_txt += f"（同比 {m['revenue_yoy']:+.1f}%" rev_txt += (f" 环比 {m['revenue_qoq']:+.1f}%）" if m.get("revenue_qoq") is not None else "）") L.append(f"【业绩概览】{rev_txt}") else: L.append("【业绩概览】营收数据缺失") if m.get("net_profit") is not None: np_txt = f"归母净利润 {m['net_profit']/1e8:.2f} 亿元" if m.get("profit_yoy") is not None: np_txt += f"（同比 {m['profit_yoy']:+.1f}%" np_txt += (f" 环比 {m['profit_qoq']:+.1f}%）" if m.get("profit_qoq") is not None else "）") L.append(f" {np_txt}") if m.get("eps") is not None: bps_txt = f"；每股净资产 {m['bps']:.2f} 元" if m.get("bps") is not None else "" L.append(f" 基本每股收益 {m['eps']:.3f} 元{bps_txt}") L.append("") L.append(f"【盈利能力】ROE {m['roe']:.1f}%　毛利率 {m['gross_margin']:.1f}%" if m.get("roe") is not None and m.get("gross_margin") is not None else "【盈利能力】数据缺失") L.append("") if highlights: L.append("✅ 亮点信号") for h in highlights: L.append(f" · {h}") if risks: L.append("⚠️ 风险提示") for r in risks: L.append(f" · {r}") L.append("") L.append(f"【综合评分】{score}/100　评级：{'🟢' if rating == '积极' else '🟡' if rating in ('中性偏积极', '中性') else '🔴'} {rating}") L.append("_" * 52) L.append("注：规则引擎自动解读，供研究参考，不构成投资建议。") return "\n".join(L) def interpret(self, periods: List[Dict]) -> Dict[str, Any]: # 统一归一化（已归一化行幂等；原始东财行/中文字段行均可） periods = [normalize_period(p) for p in periods] m = self.compute(periods) if m.get("error"): return m highlights, risks = self.signals(m) score, rating = self.score(m) return {**m, "highlights": highlights, "risks": risks, "score": score, "rating": rating, "text": self.format_text(m, highlights, risks, score, rating)} # ── 便捷入口 ────────────────────────────────────────────────── def interpret_data(data: Any) -> Dict[str, Any]: """直接喂财务数据（dict 单期 / list 多期 / 东财原始行均可）。""" if isinstance(data, dict): periods = [normalize_period(data)] elif isinstance(data, (list, tuple)): periods = [normalize_period(d) if not _is_normalized(d) else d for d in data] else: return {"error": "无法识别的数据格式"} return ReportInterpreter().interpret(periods) def _is_normalized(d: Dict) -> bool: return any(k in d for k in ("revenue", "net_profit", "eps")) def interpret_stock(stock_code: str, limit: int = 8) -> Dict[str, Any]: """一站式：拉取东财财务数据并解读。""" periods = fetch_periods(stock_code, limit=limit) if not periods: return {"error": f"未获取到 {stock_code} 的财务数据（代码可能错误或接口暂不可用）"} return ReportInterpreter().interpret(periods) def interpret_stock_report(stock_code: str, limit: int = 8) -> str: """便捷文本入口（供 MCP/CLI/对话调用）。""" result = interpret_stock(stock_code, limit=limit) return result.get("text") or f"❌ 解读失败：{result.get('error', '未知错误')}" # ── CLI ────────────────────────────────────────────────────── if __name__ == "__main__": import argparse ap = argparse.ArgumentParser(description="v7.1.0 定期报告解读引擎") ap.add_argument("stock", help="股票代码，如 600519 / 000001.SZ") ap.add_argument("--limit", type=int, default=8, help="拉取报告期数") ap.add_argument("--json", action="store_true", help="输出结构化 JSON") args = ap.parse_args() result = interpret_stock(args.stock, limit=args.limit) if args.json: print(json.dumps(result, ensure_ascii=False, indent=1, default=str)) else: print(result.get("text") or result.get("error", "解读失败"))