# -*- coding: utf-8 -*- 上市公司财务分析报告系统单文件 · 直接输出同目录 功能 1. 通过 Baostock 获取多家上市公司近 5 年年报财务数据 2. 计算盈利能力、偿债能力、运营能力、成长能力四维指标 3. 生成全部图表折线趋势 柱状对比 雷达图 4. 图表以内存流方式直接写入 Word不产生中间 PNG 文件 5. 最终只在脚本同目录输出一个 Word 文档 依赖 baostock, pandas, numpy, matplotlib, python-docx 运行 python financial_analysis.py 输出 ./财务分析报告.docx 与本脚本同目录 import os import time import io import numpy as np import pandas as pd import matplotlib.pyplot as plt from matplotlib import rcParams from docx import Document from docx.shared import Pt, Cm, RGBColor from docx.enum.text import WD_ALIGN_PARAGRAPH from docx.oxml.ns import qn import baostock as bs # # 一、配置区 # STOCK_LIST { 贵州茅台: sh.600519, 五粮液: sz.000858, 泸州老窖: sz.000568, 洋河股份: sz.002304, } MAIN_STOCK 贵州茅台 # 主分析标的 YEARS 5 # 分析年数 CURRENT_YEAR 2025 # 当前年份 # 输出与脚本同目录不新建任何文件夹 BASE_DIR os.path.dirname(os.path.abspath(__file__)) WORD_OUTPUT_PATH os.path.join(BASE_DIR, 财务分析报告.docx) # 图表样式 CHART_DPI 150 CHART_FIGSIZE (10, 5) # 中文字体按操作系统调整 # Windows: [SimHei, Microsoft YaHei] # macOS: [Arial Unicode MS] # Linux: [WenQuanYi Zen Hei] FONT_FAMILY [SimHei, Microsoft YaHei] rcParams[font.family] FONT_FAMILY rcParams[axes.unicode_minus] False # # 二、数据获取模块 # def login(): 登录 Baostock lg bs.login() print(f[Baostock] 登录状态: {lg.error_code} - {lg.error_msg}) return lg.error_code 0 def logout(): 登出 Baostock bs.logout() print([Baostock] 已登出) def _safe_get(row_data, fields, key): 安全地从 Baostock 行数据中按字段名取值 if key in fields: return row_data[fields.index(key)] return None def fetch_quarterly_data(code, year, quarter): 获取单只股票单季度的全部财务数据 row {code: code, year: year, quarter: quarter} # ---- 盈利能力 ---- rs bs.query_profit_data(codecode, yearyear, quarterquarter) while (rs.error_code 0) and rs.next(): d rs.get_row_data() f rs.fields row[statDate] _safe_get(d, f, statDate) row[毛利率] _safe_get(d, f, gpMargin) row[净利率] _safe_get(d, f, npMargin) row[ROE] _safe_get(d, f, roeAvg) # ---- 偿债能力 ---- rs bs.query_balance_data(codecode, yearyear, quarterquarter) while (rs.error_code 0) and rs.next(): d rs.get_row_data() f rs.fields row[资产负债率] _safe_get(d, f, liabilityToAsset) row[流动比率] _safe_get(d, f, currentRatio) row[速动比率] _safe_get(d, f, quickRatio) # ---- 营运能力 ---- rs bs.query_operation_data(codecode, yearyear, quarterquarter) while (rs.error_code 0) and rs.next(): d rs.get_row_data() f rs.fields row[存货周转率] _safe_get(d, f, INVTurnRatio) row[应收账款周转率] _safe_get(d, f, NRTurnRatio) row[总资产周转率] _safe_get(d, f, AssetTurnRatio) # ---- 成长能力 ---- rs bs.query_growth_data(codecode, yearyear, quarterquarter) while (rs.error_code 0) and rs.next(): d rs.get_row_data() f rs.fields row[净利润增长率] _safe_get(d, f, YOYNI) row[营收增长率] _safe_get(d, f, YOYEquity) return row def get_company_data(code, yearsYEARS): 获取某只股票近 N 年年报数据 all_rows [] for y in range(CURRENT_YEAR - years, CURRENT_YEAR): row fetch_quarterly_data(code, y, 4) if row.get(statDate): all_rows.append(row) time.sleep(0.4) # 限流 df pd.DataFrame(all_rows) if not df.empty: df[报告期] df[statDate].str[:4] 年报 numeric_cols [ 毛利率, 净利率, ROE, 资产负债率, 流动比率, 速动比率, 存货周转率, 应收账款周转率, 总资产周转率, 净利润增长率, 营收增长率, ] for c in numeric_cols: if c in df.columns: df[c] pd.to_numeric(df[c], errorscoerce) df df.sort_values(报告期).reset_index(dropTrue) return df def fetch_all(): 批量获取所有公司数据 data {} for name, code in STOCK_LIST.items(): print(f[数据] 正在获取 {name} ({code}) ...) data[name] get_company_data(code, YEARS) print(f[数据] {name}: {len(data[name])} 条记录) return data # # 三、指标计算模块 # def summarize(df: pd.DataFrame, col: str) - dict: 单列统计摘要 if df.empty or col not in df.columns: return {} s df[col].dropna() if s.empty: return {} return { 最新值: round(float(s.iloc[-1]), 4), 均值: round(float(s.mean()), 4), 最大值: round(float(s.max()), 4), 最小值: round(float(s.min()), 4), 标准差: round(float(s.std(ddof0)), 4), } def cagr(series: pd.Series, periods: int) - float: 复合增长率 CAGR s series.dropna() if len(s) 2 or periods 0: return float(nan) first, last float(s.iloc[0]), float(s.iloc[-1]) if first 0: return float(nan) return round((last / first) ** (1.0 / periods) - 1, 4) def yoy_growth(series: pd.Series) - pd.Series: 同比增速 return series.astype(float).pct_change().round(4) def score_dimension(value, best, worst, higher_betterTrue) - float: 将单指标归一化到 0-100 分 if value is None or np.isnan(value): return float(nan) if best worst: return 60.0 if higher_better: score (value - worst) / (best - worst) * 100 else: score (best - value) / (best - worst) * 100 return max(0.0, min(100.0, round(score, 2))) def composite_score(df: pd.DataFrame) - dict: 四维简化评分 if df.empty: return {} latest df.iloc[-1] def _get(col): v latest.get(col, np.nan) try: return float(v) except Exception: return np.nan roe _get(ROE) debt_ratio _get(资产负债率) asset_turn _get(总资产周转率) ni_growth _get(净利润增长率) s_profit score_dimension(roe, best0.40, worst0.05) s_debt score_dimension(debt_ratio, best0.20, worst0.70, higher_betterFalse) s_oper score_dimension(asset_turn, best1.20, worst0.20) s_growth score_dimension(ni_growth, best0.50, worst-0.20) parts [s for s in [s_profit, s_debt, s_oper, s_growth] if not np.isnan(s)] total round(sum(parts) / len(parts), 2) if parts else float(nan) return { 盈利能力得分: s_profit, 偿债能力得分: s_debt, 运营能力得分: s_oper, 成长能力得分: s_growth, 综合得分: total, } def build_summary_table(data: dict) - pd.DataFrame: 关键指标汇总表 rows [] for name, df in data.items(): if df.empty: continue rows.append({ 公司: name, 毛利率: round(df[毛利率].iloc[-1], 4) if 毛利率 in df else np.nan, 净利率: round(df[净利率].iloc[-1], 4) if 净利率 in df else np.nan, ROE: round(df[ROE].iloc[-1], 4) if ROE in df else np.nan, 资产负债率: round(df[资产负债率].iloc[-1], 4) if 资产负债率 in df else np.nan, 总资产周转率: round(df[总资产周转率].iloc[-1], 4) if 总资产周转率 in df else np.nan, 净利润增长率: round(df[净利润增长率].iloc[-1], 4) if 净利润增长率 in df else np.nan, }) return pd.DataFrame(rows) # # 四、图表生成模块全部返回内存字节流 # def _fig_to_bytes(fig) - io.BytesIO: 将 matplotlib 图形对象转换为内存字节流不落盘 buf io.BytesIO() fig.savefig(buf, formatpng, dpiCHART_DPI, bbox_inchestight) plt.close(fig) buf.seek(0) return buf def make_trend_fig(data: dict, index_name: str, title: str) - io.BytesIO: 多公司单指标趋势折线图内存流 fig, ax plt.subplots(figsizeCHART_FIGSIZE) for name in STOCK_LIST.keys(): df data.get(name) if df is None or df.empty or index_name not in df.columns: continue ax.plot(df[报告期], df[index_name], markero, labelname) ax.set_title(title) ax.set_xlabel(报告期) ax.set_ylabel(index_name) ax.legend() ax.grid(alpha0.3) return _fig_to_bytes(fig) def make_radar_fig(data: dict, company: str) - io.BytesIO: 四维雷达图内存流 df data.get(company) if df is None or df.empty: return None latest df.iloc[-1] labels [盈利能力\n(ROE), 偿债能力\n(1-资产负债率), 运营能力\n(总资产周转率), 成长能力\n(净利润增长率)] try: roe float(latest[ROE]) debt 1 - float(latest[资产负债率]) turn float(latest[总资产周转率]) grow float(latest[净利润增长率]) except Exception: return None values [max(0, roe), max(0, debt), max(0, turn), max(0, grow)] maxes [0.5, 1.0, 1.5, 0.5] values [min(v / m, 1.0) for v, m in zip(values, maxes)] angles np.linspace(0, 2 * np.pi, len(labels), endpointFalse).tolist() values values[:1] angles angles[:1] fig, ax plt.subplots(figsize(6, 6), subplot_kwdict(polarTrue)) ax.plot(angles, values, o-, linewidth2) ax.fill(angles, values, alpha0.25) ax.set_xticks(angles[:-1]) ax.set_xticklabels(labels) ax.set_ylim(0, 1) ax.set_title(f{company} 财务四维雷达图, pad20) return _fig_to_bytes(fig) def make_bar_fig(data: dict, index_name: str, title: str) - io.BytesIO: 最新一期柱状对比图内存流 names, vals [], [] for name in STOCK_LIST.keys(): df data.get(name) if df is None or df.empty or index_name not in df.columns: continue names.append(name) vals.append(df[index_name].iloc[-1]) if not names: return None fig, ax plt.subplots(figsizeCHART_FIGSIZE) bars ax.bar(names, vals, color#3498db) ax.set_title(title) ax.set_ylabel(index_name) for b, v in zip(bars, vals): ax.text(b.get_x() b.get_width() / 2, v, f{v:.2f}, hacenter, vabottom, fontsize9) ax.grid(axisy, alpha0.3) return _fig_to_bytes(fig) def generate_all_charts(data: dict) - dict: 生成全部图表返回 {逻辑名: BytesIO} charts {} # ---- 趋势图 ---- trend_metrics [ (毛利率, 毛利率 5年趋势对比), (净利率, 净利率 5年趋势对比), (ROE, ROE 5年趋势对比), (资产负债率, 资产负债率 5年趋势对比), (流动比率, 流动比率 5年趋势对比), (速动比率, 速动比率 5年趋势对比), (存货周转率, 存货周转率 5年趋势对比), (总资产周转率, 总资产周转率 5年趋势对比), (净利润增长率, 净利润增长率 5年趋势对比), ] for metric, title in trend_metrics: charts[metric _趋势] make_trend_fig(data, metric, title) print(f[图表] 已生成趋势图: {metric}) # ---- 柱状对比图 ---- bar_metrics [ (毛利率, 最新一期毛利率对比), (净利率, 最新一期净利率对比), (ROE, 最新一期ROE对比), (资产负债率, 最新一期资产负债率对比), (总资产周转率, 最新一期总资产周转率对比), ] for metric, title in bar_metrics: charts[metric _柱状] make_bar_fig(data, metric, title) print(f[图表] 已生成柱状图: {metric}) # ---- 雷达图 ---- charts[f{MAIN_STOCK}_雷达图] make_radar_fig(data, MAIN_STOCK) print(f[图表] 已生成雷达图: {MAIN_STOCK}) return charts # # 五、Word 报告生成模块 # def _set_font(run, name微软雅黑, size10.5, boldFalse, colorNone): 设置 Word 文本字体含中文 run.font.name name run.font.size Pt(size) run.font.bold bold if color: run.font.color.rgb RGBColor(*color) r run._element r.rPr.rFonts.set(qn(w:eastAsia), name) def _add_heading(doc, text, level1): 添加标题 h doc.add_heading(levellevel) run h.add_run(text) _set_font(run, size16 - level * 2, boldTrue, color(0x2C, 0x3E, 0x50)) return h def _add_para(doc, text, size10.5, boldFalse, alignNone): 添加段落 p doc.add_paragraph() run p.add_run(text) _set_font(run, sizesize, boldbold) if align: p.alignment align return p def _add_image_from_stream(doc, stream: io.BytesIO, width_cm15): 将内存中的图片字节流直接插入 Word if stream is None: return stream.seek(0) doc.add_picture(stream, widthCm(width_cm)) doc.paragraphs[-1].alignment WD_ALIGN_PARAGRAPH.CENTER def _df_to_table(doc, df: pd.DataFrame): DataFrame 写入 Word 表格 if df is None or df.empty: _add_para(doc, 暂无数据) return table doc.add_table(rows1, colslen(df.columns)) table.style Light Grid Accent 1 hdr table.rows[0].cells for i, col in enumerate(df.columns): hdr[i].text str(col) for p in hdr[i].paragraphs: for r in p.runs: _set_font(r, boldTrue) for _, row in df.iterrows(): cells table.add_row().cells for i, col in enumerate(df.columns): cells[i].text if pd.isna(row[col]) else str(row[col]) for p in cells[i].paragraphs: for r in p.runs: _set_font(r, size9) def generate_word_report(data: dict, charts: dict): 生成 Word 报告所有图表内嵌只输出一个 docx doc Document() # ---------- 封面 ---------- title doc.add_paragraph() title.alignment WD_ALIGN_PARAGRAPH.CENTER run title.add_run(上市公司财务分析报告) _set_font(run, size26, boldTrue, color(0x2C, 0x3E, 0x50)) sub doc.add_paragraph() sub.alignment WD_ALIGN_PARAGRAPH.CENTER run sub.add_run(f主分析标的{MAIN_STOCK} | 数据周期近 {YEARS} 年年报) _set_font(run, size12, color(0x55, 0x55, 0x55)) sub2 doc.add_paragraph() sub2.alignment WD_ALIGN_PARAGRAPH.CENTER run sub2.add_run(数据来源Baostock 证券宝 | 本报告仅用于课程学习不构成投资建议) _set_font(run, size10, color(0x88, 0x88, 0x88)) doc.add_page_break() # ---------- 一、总体概览 ---------- _add_heading(doc, 一、总体概览, level1) _add_para(doc, f本报告选取 {MAIN_STOCK} 及其主要同行公司 f基于 Baostock 提供的季频财务数据 f从盈利能力、偿债能力、运营能力、成长能力四个维度进行对比分析。) summary_df build_summary_table(data) _add_para(doc, 1.1 最新一期关键指标对比, boldTrue) _df_to_table(doc, summary_df) # ---------- 二、盈利能力 ---------- _add_heading(doc, 二、盈利能力分析, level1) for key in [毛利率_趋势, 净利率_趋势, ROE_趋势]: _add_image_from_stream(doc, charts.get(key)) for key in [毛利率_柱状, 净利率_柱状, ROE_柱状]: _add_image_from_stream(doc, charts.get(key), width_cm12) # ---------- 三、偿债能力 ---------- _add_heading(doc, 三、偿债能力分析, level1) for key in [资产负债率_趋势, 流动比率_趋势, 速动比率_趋势]: _add_image_from_stream(doc, charts.get(key)) _add_image_from_stream(doc, charts.get(资产负债率_柱状), width_cm12) # ---------- 四、运营能力 ---------- _add_heading(doc, 四、运营能力分析, level1) for key in [存货周转率_趋势, 总资产周转率_趋势]: _add_image_from_stream(doc, charts.get(key)) _add_image_from_stream(doc, charts.get(总资产周转率_柱状), width_cm12) # ---------- 五、成长能力 ---------- _add_heading(doc, 五、成长能力分析, level1) _add_image_from_stream(doc, charts.get(净利润增长率_趋势)) # ---------- 六、综合评分 ---------- _add_heading(doc, 六、综合评分, level1) _add_para(doc, 采用简化四维打分模型每维权重 25%满分 100 分。) score_rows [] for name, df in data.items(): sc composite_score(df) if sc: sc[公司] name score_rows.append(sc) if score_rows: score_df pd.DataFrame(score_rows)[ [公司, 盈利能力得分, 偿债能力得分, 运营能力得分, 成长能力得分, 综合得分] ] _df_to_table(doc, score_df) _add_image_from_stream(doc, charts.get(f{MAIN_STOCK}_雷达图), width_cm12) # ---------- 七、明细数据表 ---------- _add_heading(doc, f七、{MAIN_STOCK} 明细数据表, level1) main_df data.get(MAIN_STOCK) if main_df is not None and not main_df.empty: cols [报告期, 毛利率, 净利率, ROE, 资产负债率, 流动比率, 速动比率, 存货周转率, 总资产周转率, 净利润增长率] cols [c for c in cols if c in main_df.columns] _df_to_table(doc, main_df[cols]) # ---------- 八、结论 ---------- _add_heading(doc, 八、结论与建议, level1) _add_para(doc, f从近 {YEARS} 年数据来看{MAIN_STOCK} 在盈利能力方面保持行业领先 f毛利率、净利率、ROE 均处于较高水平偿债能力稳健资产负债率较低 f运营能力与成长能力需结合行业周期综合判断。 f建议结合宏观经济、行业政策、公司治理等多维度信息进行综合决策。) # ---------- 保存到脚本同目录 ---------- doc.save(WORD_OUTPUT_PATH) print(f\n[Word] 报告已生成含全部 {len(charts)} 张图表: {WORD_OUTPUT_PATH}) return WORD_OUTPUT_PATH # # 六、主程序入口 # def main(): print( * 60) print(上市公司财务分析报告系统 启动) print( * 60) if not login(): print([错误] Baostock 登录失败程序退出) return try: # 1. 拉取数据 data fetch_all() # 2. 生成全部图表内存中 print(\n[步骤] 开始生成图表内存流 ...) charts generate_all_charts(data) print(f[步骤] 共生成 {len(charts)} 张图表全部驻留内存) # 3. 生成 Word图表直接内嵌 print(\n[步骤] 开始生成 Word 报告 ...) generate_word_report(data, charts) print(\n * 60) print(全部任务完成) print(f报告路径{WORD_OUTPUT_PATH}) print( * 60) finally: logout() if __name__ __main__: main()