312 lines
8.9 KiB
Python
312 lines
8.9 KiB
Python
|
|
"""
|
|||
|
|
金融研究报告图表工具函数
|
|||
|
|
============================
|
|||
|
|
|
|||
|
|
提供符合研报规范的高度封装图表生成函数。
|
|||
|
|
|
|||
|
|
Usage:
|
|||
|
|
from chart_utils import create_research_chart, create_price_volume_chart
|
|||
|
|
|
|||
|
|
wb = create_research_chart(
|
|||
|
|
df,
|
|||
|
|
chart_type="line",
|
|||
|
|
title="股价走势",
|
|||
|
|
x_col="date",
|
|||
|
|
y_cols="close",
|
|||
|
|
output_path="report.xlsx"
|
|||
|
|
)
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
from openpyxl import Workbook
|
|||
|
|
from openpyxl.chart import LineChart, BarChart, Reference
|
|||
|
|
from openpyxl.chart.axis import DateAxis
|
|||
|
|
from openpyxl.chart.label import DataLabelList
|
|||
|
|
from openpyxl.chart.series import DataPoint, Series
|
|||
|
|
from openpyxl.chart.shapes import GraphicalProperties
|
|||
|
|
import pandas as pd
|
|||
|
|
from typing import Optional, List, Union
|
|||
|
|
|
|||
|
|
|
|||
|
|
def create_research_chart(
|
|||
|
|
df: pd.DataFrame,
|
|||
|
|
chart_type: str = "line",
|
|||
|
|
title: str = "",
|
|||
|
|
x_col: str = None,
|
|||
|
|
y_cols: Union[str, List[str]] = None,
|
|||
|
|
y_axis_title: str = "",
|
|||
|
|
x_axis_title: str = "",
|
|||
|
|
y_format: str = "0.00",
|
|||
|
|
show_data_labels: bool = False,
|
|||
|
|
show_last_label_only: bool = False,
|
|||
|
|
add_mean_line: bool = False,
|
|||
|
|
color_up: str = "FF0000",
|
|||
|
|
color_down: str = "00B050",
|
|||
|
|
tick_skip: int = 5,
|
|||
|
|
height: int = 10,
|
|||
|
|
width: int = 20,
|
|||
|
|
output_path: str = None
|
|||
|
|
) -> Workbook:
|
|||
|
|
"""
|
|||
|
|
生成符合研报规范的 Excel 图表
|
|||
|
|
|
|||
|
|
Parameters:
|
|||
|
|
-----------
|
|||
|
|
df : pd.DataFrame
|
|||
|
|
数据源,必须包含 x_col 和 y_cols 指定的列
|
|||
|
|
chart_type : str
|
|||
|
|
图表类型:"line"(折线图), "bar"(柱状图)
|
|||
|
|
title : str
|
|||
|
|
图表标题
|
|||
|
|
x_col : str
|
|||
|
|
X轴数据列名(日期/类别)
|
|||
|
|
y_cols : str or List[str]
|
|||
|
|
Y轴数据列名,支持多序列
|
|||
|
|
y_axis_title : str
|
|||
|
|
Y轴标题(建议带单位,如"价格(元)")
|
|||
|
|
x_axis_title : str
|
|||
|
|
X轴标题
|
|||
|
|
y_format : str
|
|||
|
|
Y轴数字格式,默认"0.00",百分比用"0.00%"
|
|||
|
|
show_data_labels : bool
|
|||
|
|
是否显示所有数据标签
|
|||
|
|
show_last_label_only : bool
|
|||
|
|
是否仅显示最后一个数据点的标签(标注最新值)
|
|||
|
|
add_mean_line : bool
|
|||
|
|
是否添加均值参考线
|
|||
|
|
color_up : str
|
|||
|
|
上涨/正值颜色(默认红色 FF0000)
|
|||
|
|
color_down : str
|
|||
|
|
下跌/负值颜色(默认绿色 00B050)
|
|||
|
|
tick_skip : int
|
|||
|
|
X轴标签间隔(防止重叠)
|
|||
|
|
height : int
|
|||
|
|
图表高度(厘米)
|
|||
|
|
width : int
|
|||
|
|
图表宽度(厘米)
|
|||
|
|
output_path : str
|
|||
|
|
输出文件路径,None则返回Workbook对象
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
--------
|
|||
|
|
Workbook : openpyxl Workbook对象
|
|||
|
|
|
|||
|
|
Examples:
|
|||
|
|
---------
|
|||
|
|
>>> # 股价走势图
|
|||
|
|
>>> df = pd.DataFrame({
|
|||
|
|
... 'date': ['2024-01', '2024-02', '2024-03'],
|
|||
|
|
... 'close': [10.5, 11.2, 10.8]
|
|||
|
|
... })
|
|||
|
|
>>> wb = create_research_chart(
|
|||
|
|
... df, chart_type="line",
|
|||
|
|
... title="股价走势",
|
|||
|
|
... x_col="date", y_cols="close",
|
|||
|
|
... y_axis_title="价格(元)",
|
|||
|
|
... show_last_label_only=True,
|
|||
|
|
... output_path="股价走势.xlsx"
|
|||
|
|
... )
|
|||
|
|
|
|||
|
|
>>> # 营收利润对比图
|
|||
|
|
>>> wb = create_research_chart(
|
|||
|
|
... df, chart_type="bar",
|
|||
|
|
... title="营业收入与净利润",
|
|||
|
|
... x_col="period", y_cols=["revenue", "profit"],
|
|||
|
|
... y_axis_title="金额(亿元)",
|
|||
|
|
... y_format="0.0"
|
|||
|
|
... )
|
|||
|
|
"""
|
|||
|
|
# 标准化 y_cols
|
|||
|
|
if isinstance(y_cols, str):
|
|||
|
|
y_cols = [y_cols]
|
|||
|
|
|
|||
|
|
# 创建工作簿
|
|||
|
|
wb = Workbook()
|
|||
|
|
ws = wb.active
|
|||
|
|
ws.title = "数据"
|
|||
|
|
|
|||
|
|
# 写入表头
|
|||
|
|
headers = [x_col] + y_cols
|
|||
|
|
ws.append(headers)
|
|||
|
|
|
|||
|
|
# 写入数据
|
|||
|
|
for _, row in df.iterrows():
|
|||
|
|
ws.append([row[col] for col in headers])
|
|||
|
|
|
|||
|
|
# 创建图表
|
|||
|
|
if chart_type == "line":
|
|||
|
|
chart = LineChart()
|
|||
|
|
elif chart_type == "bar":
|
|||
|
|
chart = BarChart()
|
|||
|
|
chart.type = "col"
|
|||
|
|
chart.grouping = "clustered"
|
|||
|
|
else:
|
|||
|
|
chart = LineChart()
|
|||
|
|
|
|||
|
|
# 设置标题和轴
|
|||
|
|
chart.title = title
|
|||
|
|
chart.y_axis.title = y_axis_title
|
|||
|
|
chart.x_axis.title = x_axis_title
|
|||
|
|
|
|||
|
|
# 设置图表尺寸
|
|||
|
|
chart.height = height
|
|||
|
|
chart.width = width
|
|||
|
|
|
|||
|
|
# 设置数据区域
|
|||
|
|
data_start_row = 1
|
|||
|
|
data_end_row = len(df) + 1
|
|||
|
|
|
|||
|
|
for i, y_col in enumerate(y_cols):
|
|||
|
|
col_idx = headers.index(y_col) + 1
|
|||
|
|
data_ref = Reference(ws, min_col=col_idx, min_row=data_start_row,
|
|||
|
|
max_row=data_end_row)
|
|||
|
|
cats_ref = Reference(ws, min_col=1, min_row=2, max_row=data_end_row)
|
|||
|
|
|
|||
|
|
chart.add_data(data_ref, titles_from_data=True)
|
|||
|
|
if i == 0:
|
|||
|
|
chart.set_categories(cats_ref)
|
|||
|
|
|
|||
|
|
# 设置Y轴格式
|
|||
|
|
chart.y_axis.numFmt = y_format
|
|||
|
|
|
|||
|
|
# 设置X轴标签间隔
|
|||
|
|
chart.x_axis.tickLblSkip = tick_skip
|
|||
|
|
|
|||
|
|
# 网格线设置(仅保留水平主网格线)
|
|||
|
|
chart.x_axis.majorGridlines = None
|
|||
|
|
|
|||
|
|
# 数据标签设置
|
|||
|
|
if show_data_labels:
|
|||
|
|
chart.dataLabels = DataLabelList()
|
|||
|
|
chart.dataLabels.showVal = True
|
|||
|
|
|
|||
|
|
# 仅显示最后一个标签
|
|||
|
|
if show_last_label_only and chart.series:
|
|||
|
|
series = chart.series[0]
|
|||
|
|
last_idx = len(df) - 1
|
|||
|
|
pt = DataPoint(idx=last_idx)
|
|||
|
|
pt.graphicalProperties = GraphicalProperties(solidFill=color_up)
|
|||
|
|
series.data_points = [pt]
|
|||
|
|
|
|||
|
|
# 柱状图涨跌着色
|
|||
|
|
if chart_type == "bar" and len(y_cols) == 1:
|
|||
|
|
series = chart.series[0]
|
|||
|
|
values = df[y_cols[0]].tolist()
|
|||
|
|
for i, val in enumerate(values):
|
|||
|
|
pt = DataPoint(idx=i)
|
|||
|
|
fill_color = color_up if val >= 0 else color_down
|
|||
|
|
pt.graphicalProperties = GraphicalProperties(solidFill=fill_color)
|
|||
|
|
series.data_points.append(pt)
|
|||
|
|
|
|||
|
|
# 添加均值参考线
|
|||
|
|
if add_mean_line and chart.series:
|
|||
|
|
series = chart.series[0]
|
|||
|
|
values = df[y_cols[0]].tolist()
|
|||
|
|
mean_val = sum(values) / len(values)
|
|||
|
|
|
|||
|
|
# 在工作表添加均值列
|
|||
|
|
mean_col = len(headers) + 1
|
|||
|
|
ws.cell(row=1, column=mean_col, value="均值")
|
|||
|
|
for i in range(2, data_end_row + 1):
|
|||
|
|
ws.cell(row=i, column=mean_col, value=mean_val)
|
|||
|
|
|
|||
|
|
mean_ref = Reference(ws, min_col=mean_col, min_row=1, max_row=data_end_row)
|
|||
|
|
mean_series = Series(mean_ref, title="均值")
|
|||
|
|
chart.series.append(mean_series)
|
|||
|
|
|
|||
|
|
# 图例位置(多序列时显示在底部)
|
|||
|
|
if len(y_cols) > 1 or add_mean_line:
|
|||
|
|
chart.legend.position = "b"
|
|||
|
|
else:
|
|||
|
|
chart.legend = None
|
|||
|
|
|
|||
|
|
# 添加图表到工作表
|
|||
|
|
ws.add_chart(chart, "E2")
|
|||
|
|
|
|||
|
|
# 添加数据来源注脚
|
|||
|
|
ws["A" + str(data_end_row + 2)] = "数据来源:Wind"
|
|||
|
|
|
|||
|
|
# 保存或返回
|
|||
|
|
if output_path:
|
|||
|
|
wb.save(output_path)
|
|||
|
|
|
|||
|
|
return wb
|
|||
|
|
|
|||
|
|
|
|||
|
|
def create_price_volume_chart(
|
|||
|
|
df: pd.DataFrame,
|
|||
|
|
title: str = "股价与成交量",
|
|||
|
|
date_col: str = "date",
|
|||
|
|
price_col: str = "close",
|
|||
|
|
volume_col: str = "volume",
|
|||
|
|
output_path: str = None
|
|||
|
|
) -> Workbook:
|
|||
|
|
"""
|
|||
|
|
生成股价+成交量组合图表(双Y轴)
|
|||
|
|
|
|||
|
|
Parameters:
|
|||
|
|
-----------
|
|||
|
|
df : pd.DataFrame
|
|||
|
|
包含日期、收盘价、成交量的数据
|
|||
|
|
title : str
|
|||
|
|
图表标题
|
|||
|
|
date_col, price_col, volume_col : str
|
|||
|
|
各数据列名
|
|||
|
|
output_path : str
|
|||
|
|
输出路径
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
--------
|
|||
|
|
Workbook
|
|||
|
|
"""
|
|||
|
|
wb = Workbook()
|
|||
|
|
ws = wb.active
|
|||
|
|
ws.title = "数据"
|
|||
|
|
|
|||
|
|
# 写入数据
|
|||
|
|
ws.append([date_col, price_col, volume_col])
|
|||
|
|
for _, row in df.iterrows():
|
|||
|
|
ws.append([row[date_col], row[price_col], row[volume_col]])
|
|||
|
|
|
|||
|
|
# 创建价格折线图(主Y轴)
|
|||
|
|
price_chart = LineChart()
|
|||
|
|
price_chart.title = title
|
|||
|
|
price_chart.y_axis.title = "价格(元)"
|
|||
|
|
price_chart.x_axis.title = date_col
|
|||
|
|
|
|||
|
|
price_ref = Reference(ws, min_col=2, min_row=1, max_row=len(df)+1)
|
|||
|
|
cats_ref = Reference(ws, min_col=1, min_row=2, max_row=len(df)+1)
|
|||
|
|
|
|||
|
|
price_chart.add_data(price_ref, titles_from_data=True)
|
|||
|
|
price_chart.set_categories(cats_ref)
|
|||
|
|
price_chart.y_axis.numFmt = "0.00"
|
|||
|
|
price_chart.x_axis.tickLblSkip = max(1, len(df) // 10)
|
|||
|
|
|
|||
|
|
# 创建成交量柱状图(次Y轴)
|
|||
|
|
vol_chart = BarChart()
|
|||
|
|
vol_chart.type = "col"
|
|||
|
|
vol_chart.grouping = "clustered"
|
|||
|
|
vol_chart.y_axis.axId = 200
|
|||
|
|
vol_chart.y_axis.title = "成交量(手)"
|
|||
|
|
|
|||
|
|
vol_ref = Reference(ws, min_col=3, min_row=1, max_row=len(df)+1)
|
|||
|
|
vol_chart.add_data(vol_ref, titles_from_data=True)
|
|||
|
|
|
|||
|
|
# 组合图表
|
|||
|
|
price_chart += vol_chart
|
|||
|
|
price_chart.y_axis.crosses = "max"
|
|||
|
|
|
|||
|
|
# 尺寸设置
|
|||
|
|
price_chart.height = 10
|
|||
|
|
price_chart.width = 20
|
|||
|
|
|
|||
|
|
# 隐藏网格线
|
|||
|
|
price_chart.x_axis.majorGridlines = None
|
|||
|
|
|
|||
|
|
ws.add_chart(price_chart, "E2")
|
|||
|
|
ws["A" + str(len(df) + 3)] = "数据来源:Wind"
|
|||
|
|
|
|||
|
|
if output_path:
|
|||
|
|
wb.save(output_path)
|
|||
|
|
|
|||
|
|
return wb
|