LangChain与Wind——用Mult-Agents随心所欲生成五千字股票投研报告，基于Deepseek大模型

本人持续追踪关注AI Agent与股票投资研究分析的结合，特别是Python的Crew AI库。本人所创内容均为原创，之前用的开发框架是CrewAI，现在使用了更流行的LangChain框架。目录写在前面：实现过程：最终输出**企业基本面分析报告****1. 行业地位分析****2. 估值水平评估****3. 股东结构分析****4. 分红能力推断****基本面概览结论**企业财务稳健性评级报告*

su1944

896人浏览 · 2025-04-16 20:52:31

su1944 · 2025-04-16 20:52:31 发布

本人持续追踪关注AI Agent与股票投资研究分析的结合，特别是Python的Crew AI库。本人所创内容均为原创，之前用的开发框架是Crew AI，现在使用了更流行的LangChain框架。

写在前面：

利用Mult-Agents的设计思路可见于前文，在此不再赘述：

https://blog.csdn.net/su1944/article/details/147015534?spm=1001.2014.3001.5501

LangChain是现在最流行的Agent开发框架，本人将之前利用Crew AI获得的成果更换到LangChain框架下，实现了仅需输入股票名称，即可获得报告的工作流。我已不想总结，只能三个字形容——吊爆了。

点击目录里的最终输出即可浏览成果，仅凭三到五分钟即可生成一篇五千字长文分析报告，并且数据基本准确。

话不多说，下面是代码，老规矩，想要prompt评论区或私信联系我

最终输出报告均由AI生成，用以参考前请谨慎评估，这些内容不代表作者的任何建议，投资有风险，入市需谨慎。本人不对以下内容担负任何法律责任

实现过程：

先导入WindPy，通过Wind API接入可靠的数据并通过占位符传入大语言模型Prompt是本文解决大语言模型幻觉的主要解决思路。

from WindPy import w
import pandas as pd
from datetime import datetime, timedelta

为了增强用户体验，在这我设计了仅需输入股票名即可查找所有数据函数，这样，用户可以生成任何股票的报告。

sec_name = input("请输入股票名称：")

接下来是我封装的一个将股票名称与股票代码对应，并获取所有数据的函数：

def get_all_financial_data_by_name(sec_name):
    w.start()

    # ========== 参数 ==========
    trade_date = "2025-03-30"
    report_date = "2024-09-30"
    params = f"tradeDate={trade_date};industryType=1;rptDate={report_date};currencyType=;rptType=1"

    # ========== 股票代码查找 ==========
    stock_list = w.wset("sectorconstituent", f"date={trade_date};sectorid=a001010100000000")
    df_stock_list = pd.DataFrame(stock_list.Data, index=stock_list.Fields).T
    df_stock_list.columns = stock_list.Fields

    # 根据简称查找代码
    match = df_stock_list.loc[df_stock_list["sec_name"] == sec_name]
    if match.empty:
        print(f"❌ 未找到简称为“{sec_name}”的股票")
        return None
    stock_code = match["wind_code"].values[0]
    print(f"✅ 股票简称“{sec_name}”的 Wind 代码为：{stock_code}")

    # ========== 指标字段 ==========
    basic_fields = [
        "industry_csrc12_n","mkt_cap_ard","pe_ttm","pb_mrq","pb_lyr",
        "val_dividendyield3","corpscale","nature1","majorproducttype",
        "total_shares","holder_top10pct","holder_num2","holder_avgnum","holder_pctbyssfund"
    ]
    income_fields = [
        "eps_basic", "roe_avg", "roe_basic", "roa2", "roa",
        "netprofitmargin_deducted", "grossprofitmargin", "cogstosales",
        "nptocostexpense", "expensetosales", "optoebt",
        "profittogr", "optogr", "ebittogr"
    ]
    balance_fields = [
        "debttoassets", "deducteddebttoassets2", "deducteddebttoassets",
        "longdebttolongcaptial", "longcapitaltoinvestment",
        "assetstoequity", "equity_to_asset", "catoassets",
        "currentdebttoequity", "ncatoassets", "longdebttoequity",
        "tangibleassetstoassets", "equitytototalcapital", "intdebttototalcap",
        "currentdebttodebt", "longdebtodebt", "ncatoequity", "ibdebtratio"
    ]
    cashflow_fields = [
        "salescashintoor", "ocftoor", "ocftooperateincome",
        "fa_netprofitcashcover", "capitalizedtoda", "ocftocf",
        "icftocf", "fcftocf", "ocftosales", "ocftoassets"
    ]
    price_fields = [
        "close", "open", "volume", "amt", "swing", "turn"
    ]

    # ========== 数据拉取 ==========
    def fetch_wss(fields):
        data = w.wss(stock_code, ",".join(fields), params)
        df = pd.DataFrame(data.Data, index=fields).T
        df.columns = fields
        return df

    def fetch_wsd(fields):
        data = w.wsd(stock_code, ",".join(fields), "2025-03-01", "2025-03-31", "unit=1")
        df = pd.DataFrame(data.Data, index=data.Fields).T
        df.columns = fields
        df.index = data.Times
        return df

    # 分别拉取数据
    df_basic = fetch_wss(basic_fields)
    df_income = fetch_wss(income_fields)
    df_balance = fetch_wss(balance_fields)
    df_cashflow = fetch_wss(cashflow_fields)
    df_price = fetch_wsd(price_fields)

    print("✅ 数据拉取完成")

    return {
        "code": stock_code,
        "basic": df_basic,
        "income": df_income,
        "balance": df_balance,
        "cashflow": df_cashflow,
        "price": df_price
    }

跑一下就行啦~ 这里的sec_name是用户选定的股票

data = get_all_financial_data_by_name("sec_name")

接下来老规矩，连接Deepseek的API

import getpass
import os
os.environ["DEEPSEEK_API_KEY"] = getpass.getpass()

先来调用一下试试有没有连接成功，让语言模型说自己是猫是Rain_Su（本人）惯例

from langchain_deepseek import ChatDeepSeek

model = ChatDeepSeek(
    model="deepseek-reasoner",
    temperature=0,
    max_tokens=None,
    timeout=None,
    max_retries=2,
    api_key=os.environ["DEEPSEEK_API_KEY"], 
    use_responses_api=False
)

from langchain_core.messages import HumanMessage
 
response = model.invoke(
    [HumanMessage(content="说你是猫，中文回答")]
)

print(response.content)

接下来需要对历史价格走势做一些数值处理，目前我的思路是通过字典来映射，所以Dataframe形式的多个日期的面板数据没法被传递，data_price是个dataframe，我们要提炼一下关键数据。另外我统一了一下格式，这里感谢G老师。

import numpy as np
def analyze_price_data(df: pd.DataFrame) -> dict:
    df = df.copy()
    df.index = pd.to_datetime(df.index)
    df = df.sort_index()

    # 强制数值转换
    for col in ['close', 'open', 'volume', 'amt', 'swing', 'turn']:
        df[col] = pd.to_numeric(df[col], errors='coerce')

    return {
        "start_price": round(df['close'].iloc[0], 2),
        "end_price": round(df['close'].iloc[-1], 2),
        "price_trend": round(df['close'].iloc[-1] - df['close'].iloc[0], 2),
        "max_close_date": df['close'].idxmax().strftime('%Y-%m-%d'),
        "max_close_price": round(df['close'].max(), 2),
        "min_close_date": df['close'].idxmin().strftime('%Y-%m-%d'),
        "min_close_price": round(df['close'].min(), 2),
        "price_change_percent": round((df['close'].iloc[-1] - df['close'].iloc[0]) / df['close'].iloc[0] * 100, 2),
        "max_swing_date": df['swing'].idxmax().strftime('%Y-%m-%d'),
        "max_swing": round(df['swing'].max(), 2),
        "max_volume_date": df['volume'].idxmax().strftime('%Y-%m-%d'),
        "max_volume": int(df['volume'].max()),
        "max_amt": round(df['amt'].max() / 1e8, 2),
        "avg_turnover": round(df['turn'].mean(), 2),
        "high_volatility_days": int((df['swing'] > 3).sum()),
        "critical_support": round(df['close'].quantile(0.1), 2),
    }

def ensure_lowercase_keys(data):
    if isinstance(data, dict) and "Fields" in data and "Data" in data:
        # WindPy 风格数据
        return {
            field.lower(): data["Data"][i][0] if data["Data"][i] else None
            for i, field in enumerate(data["Fields"])
        }
    elif isinstance(data, pd.DataFrame):
        # 已是 DataFrame，返回列名小写版
        data.columns = [col.lower() for col in data.columns]
        return data
    else:
        raise TypeError("Unsupported data format passed to ensure_lowercase_keys")


def to_builtin_types(d):
    return {
        k: float(v) if isinstance(v, (np.float64, np.float32)) else
           int(v) if isinstance(v, (np.int64, np.int32)) else
           str(v) if isinstance(v, (np.datetime64, pd.Timestamp)) else
           v
        for k, v in d.items()
    }

转换成字典，统一格式：

basic_data = data["basic"].iloc[0].to_dict()
balance_data = data["balance"].iloc[0].to_dict()
income_data = data["income"].iloc[0].to_dict()
cashflow_data = data["cashflow"].iloc[0].to_dict()
price_data = analyze_price_data(data["price"])
price_data = to_builtin_types(price_data)

接下来利用LangChain搭建Mult- Agent 分析框架，这里我们直接用了Deepseek-reasoner，也就是Deepseek R1，用以确保结果的可靠性。

from typing import TypedDict
from langgraph.graph import StateGraph, END
from langchain_core.runnables import RunnableLambda
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
import matplotlib.pyplot as plt
from langgraph.graph import START

llm = ChatDeepSeek(model="deepseek-reasoner")

def make_analysis_node(template: str):
    prompt = PromptTemplate.from_template(template)
    return prompt | llm | StrOutputParser()

# Define nodes
basic_node = make_analysis_node(analyze_basic_template)
balance_node = make_analysis_node(analyze_balance_template)
income_node = make_analysis_node(analyze_income_template)
cashflow_node = make_analysis_node(analyze_cash_template)
price_node = make_analysis_node(analyze_price_template)
synthesis_node = make_analysis_node(synthesis_template)

# Define update wrapper
def run_and_store(field_in, field_out, node):
    return RunnableLambda(lambda x: {field_out: node.invoke(x[field_in])})

# Build LangGraph
class ResearchState(TypedDict):
    basic_data: str
    balance_data: str
    income_data: str
    cashflow_data: str
    price_data: str
    basic_analysis: str
    balance_analysis: str
    income_analysis: str
    cashflow_analysis: str
    price_analysis: str
    report: str

builder = StateGraph(ResearchState)
builder.add_node("AnalyzeBasic", run_and_store("basic_data", "basic_analysis", basic_node))
builder.add_node("AnalyzeBalance", run_and_store("balance_data", "balance_analysis", balance_node))
builder.add_node("AnalyzeIncome", run_and_store("income_data", "income_analysis", income_node))
builder.add_node("AnalyzeCashflow", run_and_store("cashflow_data", "cashflow_analysis", cashflow_node))
builder.add_node("AnalyzePrice", run_and_store("price_data", "price_analysis", price_node))

# 添加边：从 START 启动所有分析节点（并行执行）
builder.add_edge(START, "AnalyzeBasic")
builder.add_edge(START, "AnalyzeBalance")
builder.add_edge(START, "AnalyzeIncome")
builder.add_edge(START, "AnalyzeCashflow")
builder.add_edge(START, "AnalyzePrice")

builder.add_node("Synthesize", RunnableLambda(lambda x: {
    "report": synthesis_node.invoke({
        "basic_analysis": x["basic_analysis"],
        "balance_analysis": x["balance_analysis"],
        "income_analysis": x["income_analysis"],
        "cashflow_analysis": x["cashflow_analysis"],
        "price_analysis": x["price_analysis"]
    })
}))

builder.add_edge("AnalyzeBasic", "Synthesize")
builder.add_edge("AnalyzeBalance", "Synthesize")
builder.add_edge("AnalyzeIncome", "Synthesize")
builder.add_edge("AnalyzeCashflow", "Synthesize")
builder.add_edge("AnalyzePrice", "Synthesize")

builder.add_edge("Synthesize", END)

graph = builder.compile()

定义一个运行函数：

def run_investment_analysis(basic_data, balance_data, income_data, cashflow_data, price_data):
    inputs = {
        "basic_data": basic_data,
        "balance_data": balance_data,
        "income_data": income_data,
        "cashflow_data": cashflow_data,
        "price_data": price_data
    }
    result = graph.invoke(inputs)

    return {
        "sub_analyses": {
            "basic": result["basic_analysis"],
            "balance": result["balance_analysis"],
            "income": result["income_analysis"],
            "cashflow": result["cashflow_analysis"],
            "price": result["price_analysis"]
        }
    }

执行，用的Deepseek- R1，所以生成会很慢。

result = run_investment_analysis(basic_data, balance_data, income_data, cashflow_data, price_data)

在这里我们把储存在result里的结果格式化成一份word版的投研文字报告。

from docx import Document
from docx.shared import Pt
from docx.oxml.ns import qn

import re

def add_markdown_to_doc(doc, md_text):
    lines = md_text.split('\n')
    for line in lines:
        line = line.strip()
        if not line:
            doc.add_paragraph()
            continue

        # 处理标题
        if line.startswith('### '):
            para = doc.add_paragraph(line[4:], style='Heading 2')
        elif line.startswith('#### '):
            para = doc.add_paragraph(line[5:], style='Heading 3')
        # 处理无序列表
        elif line.startswith('- ') or line.startswith('* '):
            para = doc.add_paragraph(line[2:], style='List Bullet')
        # 处理加粗（**text**）
        elif '**' in line:
            para = doc.add_paragraph()
            while '**' in line:
                parts = re.split(r'(\*\*.*?\*\*)', line, maxsplit=1)
                for part in parts:
                    if part.startswith('**') and part.endswith('**'):
                        run = para.add_run(part[2:-2])
                        run.bold = True
                        run.font.name = 'SimHei'
                        run._element.rPr.rFonts.set(qn('w:eastAsia'), '黑体')
                    else:
                        run = para.add_run(part)
                        run.font.name = 'SimHei'
                        run._element.rPr.rFonts.set(qn('w:eastAsia'), '黑体')
                break
        else:
            para = doc.add_paragraph(line)
        
        # 设置字体
        for run in para.runs:
            run.font.name = 'SimHei'
            run.font.size = Pt(11)
            run._element.rPr.rFonts.set(qn('w:eastAsia'), '黑体')
            
doc = Document()
doc.add_heading('企业财务分析报告', 0)
doc.paragraphs[-1].runs[0].font.name = 'SimHei'
doc.paragraphs[-1].runs[0]._element.rPr.rFonts.set(qn('w:eastAsia'), '黑体')

# 逐部分写入并渲染 Markdown
for section in result['sub_analyses'].values():
    add_markdown_to_doc(doc, section)

doc.save("企业财务分析报告.docx")

最终输出

企业基本面分析报告

---

1. 行业地位分析

**所属行业特征**：制造业（专用设备与零部件）通常具有较高的技术壁垒和周期性特征，细分领域龙头企业的护城河较强，但行业竞争激烈，对研发投入和供应链管理要求较高。
**企业规模与定位**：
**大型民营企业**：市值超1.13万亿元，属于行业头部企业，规模效应显著，具备成本控制、议价能力和资源整合优势。
**产品聚焦**：主营专用设备与零部件，若产品覆盖高端制造（如半导体设备、新能源装备等），则行业地位可能进一步强化；但需警惕下游需求波动风险（如基建、工业投资周期）。
**竞争格局**：需结合市占率数据进一步验证，但大型规模通常对应较强的行业话语权。

---

2. 估值水平评估

**市盈率（TTM 22.28）**：
高于A股制造业平均市盈率（约15-20倍），反映市场对其盈利成长性或行业稀缺性存在溢价预期；需结合净利润增速判断合理性（若增速低于20%，则估值偏高）。
**市净率（MRQ 4.58）**：
显著高于制造业平均市净率（通常1-3倍），可能表明：

（1）资产轻量化（如技术/品牌溢价未体现在账面）；

（2）高ROE（需验证净资产盈利能力）；

（3）市场情绪推动估值泡沫。

**历史对比**：当前市净率与历史值（4.58）基本持平，未出现明显偏离，但需警惕高估值下的回调风险。

---

3. 股东结构分析

**股权集中度**：前十大股东持股62.27%，属于**中度集中型**结构，利于战略决策效率，但需关注大股东质押或减持风险。
**散户与机构持仓**：
**社保基金持仓缺失**：若长期无社保/公募持仓，可能隐含机构对其长期价值或流动性的担忧。
**股东人数数据缺失**：无法判断筹码分散度（散户参与度），需补充数据进一步分析。
**潜在风险**：民营企业可能存在实控人“一言堂”治理风险，需关注关联交易与治理透明度。

---

4. 分红能力推断

**盈利支撑**：若市盈率22倍对应合理净利润（市值/PE），则年净利润约507亿元，理论上具备高分红潜力。
**分红意愿**：
民企分红政策通常灵活，但需平衡扩张需求（专用设备行业依赖持续研发投入）；
若无稳定分红历史或高负债率，则实际分红率可能受限。
**现金流验证**：需结合自由现金流数据（未提供）判断分红可持续性，警惕“纸面利润”陷阱。

---

基本面概览结论

1. 行业地位：**头部企业，具备规模与技术壁垒**，但需关注行业周期性与下游需求韧性。

2. 估值水平：**偏高，依赖成长预期支撑**，若业绩增速不达预期或行业景气度下行，存在估值修正压力。

3. 股东结构：**控股权集中，机构参与度存疑**，需警惕治理风险与流动性问题。

4. 分红能力：**理论上有潜力，但需验证盈利质量与现金流**，可能优先投入再生产而非高分红。

综合评级：**具备行业龙头属性，但高估值需业绩高增长兑现支撑，短期更适合风险偏好较高的投资者**。建议补充毛利率、ROE、负债率等数据完善分析。

---

注：部分关键数据缺失（如股东人数、社保持仓、现金流），结论存在一定假设性。

企业财务稳健性评级报告

---

一、偿债能力分析

1. 资产负债率（64.33%）

该比率高于50%警戒线，但低于70%高风险阈值，表明企业有一定债务压力，但尚未达到高风险水平。需结合行业特性判断（如重资产行业可能容忍更高负债率）。
**风险点**：若资产流动性不足，可能面临偿债压力。

2. 有息负债占比（17.55%）

有息负债比例较低，说明企业依赖外部融资的程度适中，利息负担可控。

3. 债务期限结构（短期债务59.18% vs 长期债务40.82%）

短期债务占比较高，需关注流动性风险（如流动负债与权益比达118.6%），但货币资金占总资产63.23%可部分对冲此风险。

偿债能力评级：★★★☆（中等偏上）

短期风险可控（货币资金充足），但长期需优化债务期限结构。

---

二、资本结构评估

1. 股东权益占比（60.31%）

股东权益占主导地位，资本结构较为稳健，抗风险能力较强。

2. 长期负债/资本比（45.00%）

长期融资占比偏高，需关注长期债务的利息成本与资产回报率匹配性。

3. 权益乘数（2.80）与股东权益/资产比（35.67%）

权益乘数较高（＞2），表明企业通过杠杆放大收益，但需警惕市场波动或利润下滑时的偿债压力。

资本结构评级：★★★（中等）

杠杆使用激进，但股东权益基础稳固，需平衡杠杆与风险。

---

三、资产结构与质量

1. 非流动资产比重（36.77%）

轻资产运营特征显著，资产流动性强（流动资产占比63.23%），但需关注流动资产质量（如应收款账期、存货周转率）。

2. 流动负债与权益比（118.62%）

流动负债超过股东权益，短期偿债压力显著，但高货币资金（63.23%）提供强流动性保障。

3. 货币资金占比（63.23%）

现金储备异常充裕，可能反映企业保守的财务策略或短期投资需求，需验证是否存在大额受限资金。

资产质量评级：★★★★（良好）

流动性极强，但需优化资金使用效率。

---

四、财务稳健性综合评级

| 维度 | 评分（5★制） | 关键结论 |

|--------------|-------------|------------------------------|

| 偿债能力 | ★★★☆ | 短期无忧，长期需优化债务结构 |

| 资本结构 | ★★★ | 杠杆偏高但权益基础稳固 |

| 资产质量 | ★★★★ | 流动性强，资金效率待提升 |

综合评级：★★★☆（中等偏上）

企业财务整体稳健，短期流动性充足，但需降低资产负债率、优化债务期限结构以提高长期抗风险能力。若货币资金未受限且经营现金流稳定，可上调至★★★★。

---

改进建议

1. 降低短期债务依赖：逐步置换短期债务为长期低成本融资。

2. 提升资金效率：盘活冗余货币资金，用于研发、投资或股东回报。

3. 监控流动性风险：尽管现金充足，仍需避免流动负债进一步攀升。

企业利润表综合分析及经营评级

---

一、盈利能力分析

1. 每股收益（EPS）

EPS为 8.1894，表明公司每股盈利能力强劲。若行业平均EPS较低（如低于5），则公司处于行业领先地位；若行业EPS普遍较高（如科技或高增长行业），需结合估值倍数进一步判断竞争力。

2. 股东回报与资产效率

**ROE（16.57%）**：显著高于10%的基准值，反映股东资金利用效率优异，可能受益于高净利率或财务杠杆放大效应。
**ROA（5.32%）**：与ROE差距较大，暗示公司可能通过较高负债（如资产负债率超50%）提升ROE，需警惕财务风险。

3. 净利润率（14.95%）

每百元收入留存近15元净利润，显著优于多数制造业（5-10%）和零售业（2-5%），接近科技或消费品牌企业水平，体现较强的议价能力和费用管控能力。

---

二、成本与费用控制

1. 毛利率（28.19%）与成本率（71.81%）

毛利率偏低，接近传统制造业（20-30%）或零售业（如超市），可能面临成本压力。需结合行业判断：若行业毛利率均值高于30%（如消费品），则存在优化空间；若行业均值相近（如汽车制造），则属正常水平。
高销售成本率指向原材料、人工或物流成本压力，需通过供应链优化或技术升级改善。

2. 费用率（10.76%）

销售、管理及财务费用占收入10.76%，优于多数企业（如零售业通常15-20%），表明公司在市场推广、行政管理等环节效率较高。

3. 净利/成本费用比（18.11%）

每投入100元成本费用可产出18.11元净利润，覆盖能力较强，验证了费用控制的有效性。

---

三、利润质量与结构

1. 扣非净利率（12.42%）

与净利润率（14.95%）差异较小，说明利润主要来源于主营业务，非经常性损益（如资产处置、政府补贴）占比较低，利润质量较高。

2. EBIT与营业利润率（16.04% vs 17.81%）

营业利润率高于EBIT率，可能因利息收入或非核心业务收益正向贡献。
两项指标均超过15%，凸显核心业务盈利能力的稳健性，优于多数实体企业（如制造业EBIT率通常5-10%）。

3. 营业利润/利润总额（100.89%）

利润几乎全部来自营业活动，非经营性收益（如投资收益）占比微乎其微，利润结构健康，可持续性强。

---

四、经营表现综合评级

综合评级：A-

**优势**：高ROE与净利润率、优异的费用控制能力、利润结构健康。
**风险**：毛利率偏低、ROA与ROE差距隐含潜在财务杠杆风险。
**建议**：

1. 优化供应链或提升产品附加值以提高毛利率；

2. 审慎管理负债结构，避免杠杆过高导致偿债压力；

3. 维持现有费用控制策略，巩固核心业务优势。

注：需结合行业数据（如毛利率/ROE行业均值）及非财务指标（市场份额、研发投入）进一步验证结论。

企业现金流量表深度解析与财务健康评级

---

一、主营业务现金创造能力

1. 销售回款率（122.58%）

**评价**：销售回款率显著高于100%，表明公司销售变现能力极强，可能通过预收账款、缩短信用周期或加速应收账款回收实现。
**风险点**：若行业普遍采用赊销模式，高回款率可能反映公司收紧信用政策，需警惕未来销售增长受限或客户流失风险。

2. OCF/营业收入（26.04%）

**评价**：经营性现金流（OCF）占收入比例较高（>20%），表明主营业务的现金转化效率良好，利润质量较高。
**行业对比**：若为制造业或消费行业，该水平属优秀；若为高研发行业（如科技），需结合研发支出现金流进一步分析。

3. OCF/营业利润（187.19%）

**评价**：经营性现金流远超营业利润，反映非付现成本（如折旧摊销）对利润的稀释，或存在营业外收入虚增利润的情况。需关注扣非净利润的真实性。

4. 扣非净利润现金保障倍数（187.34%）

**评价**：扣非净利润完全由现金支撑，盈利质量极高，企业核心业务具备可持续造血能力。

评级：**优秀**（现金创造能力强劲，盈利质量可靠）

---

二、现金流结构与稳定性

1. 现金流结构异常

**经营现金流占比（-7125.1%）**：极端负值表明经营现金流净流出，或分母（三项现金流总和）因其他活动现金大幅流入导致结构失衡。需结合绝对数值验证（可能数据口径问题）。
**投资现金流占比（5059.8%）**：显著主导地位，可能因大规模资产处置（如出售子公司、固定资产）或收回投资（如理财产品到期），需警惕核心资产流失风险。
**自由现金流占比（2165.3%）**：自由现金流（OCF - 资本开支）为正且占比高，表明公司具备持续投资能力，但需结合投资方向合理性判断。

2. OCF/销售收入（26.04%）与OCF/总资产（9.14%）