可视化功能概述
本方案面向“近三个月的月度收支”数据，提供一套可扩展的堆叠面积图可视化模块，支持：
- 收入（工资、副业）与支出（房租、餐饮、出行）的分项堆叠展示，收入为正向堆叠、支出为负向堆叠，直观体现收支结构。
- 自动计算每月结余与结余率，结余以折线显示，便于识别盈余变化。
- 标注异常波动（支持百分比变化阈值或z-score两种方法）与工资发放日（垂直虚线+标签）。
- 配色与堆叠配置建议，保证信息清晰、对比鲜明、避免误导。
- 提供Plotly交互式图表（默认）以及Matplotlib本地静态图表（可选）。
注意：所有图表仅在传入真实的用户数据时生成，示例中不包含虚构数据；请按示例接口传入您自己的CSV/DataFrame数据。
核心接口定义
from typing import List, Dict, Optional, Tuple, Union
import pandas as pd
import numpy as np
try:
import plotly.graph_objects as go
except ImportError:
go = None
def detect_anomalies(
series: pd.Series,
method: str = "pct_change",
threshold: float = 0.25
) -> List[int]:
"""
异常检测：
- method='pct_change': 月环比变化绝对值 >= threshold（如0.25代表±25%）标记为异常
- method='zscore': 计算z分数，|z| >= threshold（如1.5）标记为异常
返回异常点的索引列表（相对series的顺序索引）
"""
series = series.astype(float)
if method == "pct_change":
pct = series.pct_change()
return [i for i, v in enumerate(pct) if i > 0 and pd.notnull(v) and abs(v) >= threshold]
elif method == "zscore":
mu = series.mean()
sigma = series.std(ddof=1)
if sigma == 0 or np.isnan(sigma):
return []
z = (series - mu) / sigma
return [i for i, v in enumerate(z) if abs(v) >= threshold]
else:
raise ValueError("Unsupported anomaly method. Use 'pct_change' or 'zscore'.")
def monthly_cashflow_area(
df: pd.DataFrame,
income_cols: List[str],
expense_cols: List[str],
month_col: str = "month",
salary_paydays: Optional[Dict[str, str]] = None,
anomaly_on: str = "total", # 'total' 或 'surplus_rate'
anomaly_method: str = "pct_change", # 'pct_change' 或 'zscore'
anomaly_threshold: float = 0.25,
colors: Optional[Dict[str, str]] = None,
library: str = "plotly", # 'plotly' 或 'matplotlib'
title: Optional[str] = None,
return_components: bool = False
) -> Union["go.Figure", "matplotlib.figure.Figure", Tuple[object, pd.DataFrame, Dict]]:
"""
生成“近三个月月度收支”堆叠面积图，并计算结余与结余率、标注异常与工资发放日。
参数：
- df: 包含月份与各项收支的DataFrame。必须包含month_col、income_cols、expense_cols对应的列。
- income_cols: 收入列名列表，例如 ['income_salary', 'income_side']。
- expense_cols: 支出列名列表，例如 ['expense_rent', 'expense_dining', 'expense_transport']。
- month_col: 月份列，建议为可解析为日期的字符串（如'2025-09-01'）或datetime。
- salary_paydays: 可选，字典 { 'YYYY-MM': 'YYYY-MM-DD' }，用于在图中标记工资发放日。
- anomaly_on: 异常检测维度，'total'在总收入与总支出上检测，'surplus_rate'在结余率上检测。
- anomaly_method: 异常检测方法，'pct_change'（环比百分比变化）或 'zscore'（z分数）。
- anomaly_threshold: 异常阈值，pct_change时如0.25=±25%，zscore时如1.5=|z|≥1.5。
- colors: 可选，颜色映射字典。未提供时使用推荐配色。
- library: 可视化库，'plotly'（默认，交互）或 'matplotlib'（静态）。
- title: 图表标题。
- return_components: True时返回(figure, metrics_df, diagnostics)。
返回：
- figure: Plotly或Matplotlib图对象。
- （可选）metrics_df: 含 total_income、total_expense、surplus、surplus_rate 的DataFrame。
- （可选）diagnostics: 包含异常索引与文本的字典。
"""
# 校验列存在
for col in [month_col] + income_cols + expense_cols:
if col not in df.columns:
raise ValueError(f"DataFrame缺少必要列: {col}")
# 复制并预处理
work = df.copy()
# 规范月份为datetime，便于排序和轴显示
work[month_col] = pd.to_datetime(work[month_col])
work.sort_values(by=month_col, inplace=True)
work.reset_index(drop=True, inplace=True)
# 计算总计、结余与结余率
work["total_income"] = work[income_cols].astype(float).sum(axis=1)
work["total_expense"] = work[expense_cols].astype(float).sum(axis=1)
work["surplus"] = work["total_income"] - work["total_expense"]
work["surplus_rate"] = np.where(
work["total_income"] > 0,
work["surplus"] / work["total_income"],
np.nan
)
# 异常检测
if anomaly_on == "total":
inc_anom_idx = detect_anomalies(work["total_income"], method=anomaly_method, threshold=anomaly_threshold)
exp_anom_idx = detect_anomalies(work["total_expense"], method=anomaly_method, threshold=anomaly_threshold)
anomalies = {"income": inc_anom_idx, "expense": exp_anom_idx}
elif anomaly_on == "surplus_rate":
sr_anom_idx = detect_anomalies(work["surplus_rate"], method=anomaly_method, threshold=anomaly_threshold)
anomalies = {"surplus_rate": sr_anom_idx}
else:
raise ValueError("anomaly_on 必须为 'total' 或 'surplus_rate'")
# 推荐配色（可覆盖）
default_colors = {
# 收入
"income_salary": "#2ecc71", # 绿 - 工资
"income_side": "#27ae60", # 深绿 - 副业
# 支出
"expense_rent": "#e74c3c", # 红 - 房租
"expense_dining": "#f39c12", # 橙 - 餐饮
"expense_transport": "#e67e22", # 橘 - 出行
# 结余
"surplus": "#34495e" # 蓝灰 - 结余线
}
# 合并用户配色
palette = {**default_colors, **(colors or {})}
# 工资发放日解析为datetime
payday_map = {}
if salary_paydays:
for k, v in salary_paydays.items():
try:
payday_map[pd.to_datetime(k).strftime("%Y-%m")] = pd.to_datetime(v)
except Exception as e:
raise ValueError(f"工资发放日解析失败: {k} -> {v}. {e}")
# 构图
if library == "plotly":
if go is None:
raise ImportError("未安装 plotly，请选择 library='matplotlib' 或安装 plotly")
fig = go.Figure()
x = work[month_col]
# 收入堆叠（正向）
for col in income_cols:
fig.add_trace(go.Scatter(
x=x, y=work[col], name=col,
stackgroup="income", mode="lines",
line={"width": 1, "color": palette.get(col, "#2ecc71")},
fill="tonexty",
hovertemplate=f"%{{x|%Y-%m}}<br>{col}: %{{y:,.2f}}<extra></extra>"
))
# 支出堆叠（负向）
for col in expense_cols:
fig.add_trace(go.Scatter(
x=x, y=-work[col], name=col,
stackgroup="expense", mode="lines",
line={"width": 1, "color": palette.get(col, "#e74c3c")},
fill="tonexty",
hovertemplate=f"%{{x|%Y-%m}}<br>{col}: -%{{y:,.2f}}<extra></extra>"
))
# 结余线
fig.add_trace(go.Scatter(
x=x, y=work["surplus"], name="结余",
mode="lines+markers",
line={"width": 2, "color": palette.get("surplus", "#34495e")},
hovertemplate="%{x|%Y-%m}<br>结余: %{y:,.2f}<br>结余率: %{customdata:.1%}<extra></extra>",
customdata=work["surplus_rate"]
))
# 工资发放日标注（垂直虚线 + 文本）
if payday_map:
xmin, xmax = x.min(), x.max()
for m, d in payday_map.items():
if d >= xmin and d <= xmax:
fig.add_vline(
x=d, line_width=1, line_dash="dot", line_color="#7f8c8d"
)
fig.add_annotation(
x=d, y=max(work["total_income"].max(), work["surplus"].max()),
text="工资日", showarrow=False, yshift=10,
font={"size": 10, "color": "#7f8c8d"}
)
# 异常标注
def _annotate(idx: int, text: str, color: str = "#c0392b"):
fig.add_annotation(
x=work.loc[idx, month_col],
y=work.loc[idx, "surplus"],
text=text,
showarrow=True, arrowhead=2, ax=0, ay=-30,
font={"size": 10, "color": color},
bgcolor="rgba(255,255,255,0.7)"
)
if "income" in anomalies:
for i in anomalies["income"]:
_annotate(i, "收入异常")
if "expense" in anomalies:
for i in anomalies["expense"]:
_annotate(i, "支出异常")
if "surplus_rate" in anomalies:
for i in anomalies["surplus_rate"]:
_annotate(i, "结余率异常")
fig.update_layout(
title=title or "近三个月收支堆叠面积图（含结余与标注）",
hovermode="x unified",
legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1),
margin=dict(l=40, r=20, t=60, b=40),
template="plotly_white"
)
fig.update_yaxes(
title="金额（收入为正，支出为负）",
zeroline=True, zerolinewidth=1, zerolinecolor="#95a5a6"
)
fig.update_xaxes(
dtick="M1", tickformat="%Y-%m", showgrid=True
)
elif library == "matplotlib":
import matplotlib.pyplot as plt
fig, ax = plt.subplots(figsize=(9, 5))
x = work[month_col].dt.to_pydatetime()
# 收入堆叠
income_data = [work[col].values for col in income_cols]
ax.stackplot(x, *income_data, labels=income_cols,
colors=[palette.get(c, "#2ecc71") for c in income_cols], alpha=0.85)
# 支出堆叠（负值）
expense_data = [(-work[col].values) for col in expense_cols]
ax.stackplot(x, *expense_data, labels=expense_cols,
colors=[palette.get(c, "#e74c3c") for c in expense_cols], alpha=0.85)
# 结余线
ax.plot(x, work["surplus"], label="结余", color=palette.get("surplus", "#34495e"), linewidth=2, marker="o")
# 工资日
if payday_map:
for m, d in payday_map.items():
if d >= work[month_col].min() and d <= work[month_col].max():
ax.axvline(d, color="#7f8c8d", linestyle=":", linewidth=1)
ax.text(d, work["surplus"].max(), "工资日", color="#7f8c8d", fontsize=9, va="bottom")
# 异常标注
def _annotate(idx: int, text: str, color: str = "#c0392b"):
ax.annotate(text, xy=(x[idx], work.loc[idx, "surplus"]),
xytext=(0, -25), textcoords="offset points",
arrowprops=dict(arrowstyle="->", color=color),
fontsize=9, color=color,
bbox=dict(boxstyle="round,pad=0.2", fc="white", alpha=0.7))
if "income" in anomalies:
for i in anomalies["income"]:
_annotate(i, "收入异常")
if "expense" in anomalies:
for i in anomalies["expense"]:
_annotate(i, "支出异常")
if "surplus_rate" in anomalies:
for i in anomalies["surplus_rate"]:
_annotate(i, "结余率异常")
ax.set_title(title or "近三个月收支堆叠面积图（含结余与标注）")
ax.set_ylabel("金额（收入为正，支出为负）")
ax.grid(True, axis="y", linestyle="--", alpha=0.3)
ax.legend(loc="upper center", ncol=3, bbox_to_anchor=(0.5, 1.12))
else:
raise ValueError("library 必须为 'plotly' 或 'matplotlib'")
diagnostics = {
"anomalies": anomalies,
"paydays": payday_map
}
metrics_df = work[[month_col, "total_income", "total_expense", "surplus", "surplus_rate"]].copy()
if return_components:
return fig, metrics_df, diagnostics
return fig
使用示例
注意：为遵循“所有图表必须基于真实数据”的规则，以下示例只展示如何加载与调用。请用您真实的近三个月数据文件或DataFrame替换示例中的路径或变量。
示例一：基础用法（Plotly，CSV加载）
import pandas as pd
# 请准备真实数据CSV，列至少包含：
# month, income_salary, income_side, expense_rent, expense_dining, expense_transport
df = pd.read_csv("your_finance_3m.csv")
fig = monthly_cashflow_area(
df=df,
income_cols=["income_salary", "income_side"],
expense_cols=["expense_rent", "expense_dining", "expense_transport"],
month_col="month",
library="plotly",
title="近三个月收支堆叠面积图"
)
fig.show()
示例二：标注工资发放日 + 自定义异常检测与配色
import pandas as pd
df = pd.read_csv("your_finance_3m.csv")
custom_colors = {
"income_salary": "#2ecc71",
"income_side": "#1abc9c",
"expense_rent": "#e74c3c",
"expense_dining": "#f39c12",
"expense_transport": "#d35400",
"surplus": "#34495e"
}
# 将每个月映射到实际工资发放日（请填入真实日期）
salary_paydays = {
"2025-08": "2025-08-25",
"2025-09": "2025-09-25",
"2025-10": "2025-10-25"
}
fig, metrics, diag = monthly_cashflow_area(
df=df,
income_cols=["income_salary", "income_side"],
expense_cols=["expense_rent", "expense_dining", "expense_transport"],
month_col="month",
salary_paydays=salary_paydays,
anomaly_on="total",
anomaly_method="pct_change",
anomaly_threshold=0.25, # 环比变化超过±25%标记为异常
colors=custom_colors,
library="plotly",
title="近三个月收支堆叠面积图（工资日与异常标注）",
return_components=True
)
print("指标：")
print(metrics) # 查看每月总收入/总支出/结余/结余率
print("诊断：", diag) # 查看异常索引、工资日字典
fig.show()
示例三：Matplotlib静态图（适合离线报告）
import pandas as pd
df = pd.read_csv("your_finance_3m.csv")
fig = monthly_cashflow_area(
df=df,
income_cols=["income_salary", "income_side"],
expense_cols=["expense_rent", "expense_dining", "expense_transport"],
month_col="month",
library="matplotlib",
title="近三个月收支堆叠面积图（静态）"
)
import matplotlib.pyplot as plt
plt.show()
图表效果描述
- 视觉编码：
- 收入为正向堆叠面积（绿色系），支出为负向堆叠面积（红橙系），中轴为零线。通过上下堆叠形成“收支对比”，可一眼看出各类别占比与总额差异。
- 结余使用深蓝灰折线+标记点展示，放在同一坐标系，便于观察盈余随月变化。
- 交互提示（Plotly）：
- 鼠标悬停显示月份、类别与金额，收入显示正值，支出显示带负号的数值（强调支出方向）。
- 统一的hover模式（x unified）在同一竖直线同时显示各项。
- 异常标注：
- 默认在总收入与总支出上检测环比变化超过±25%的月份，并在结余线附近添加“收入异常/支出异常”提示。
- 可切换为对“结余率”做z-score检测（如阈值1.5），适合数据稳定但存在异常波动的场景。
- 工资发放日：
- 在实际工资发放日期处绘制垂直虚线，并在顶部添加“工资日”标注，帮助关联当月现金流峰值与薪资入账时点。
- 堆叠面积图配置建议：
- Stack分组：收入与支出分别使用不同stackgroup，支出以负值显示，避免误导总额。
- 透明度：面积图填充透明度建议在0.8左右，既突出又不过度遮挡。
- 网格与零线：保留y轴零线与浅色网格，帮助识别正负区域与变化幅度。
- 轴刻度：x轴按月份显示（%Y-%m），保证时间序可读；y轴单位统一（如元或千元），必要时添加千分位格式。
扩展建议
- 趋势分析增强：
- 引入三期移动平均（3M MA）或指数平滑，辅助识别收支趋势（不替代原值，仅作为辅线）。
- 增加同比（YoY）与环比（MoM）标签，在hover中显示变化比例。
- 异常识别个性化：
- 分类别异常阈值（如房租一般稳定、餐饮和出行波动较大），支持为不同列设置差异化阈值。
- 将异常划分为“高于预期/低于预期”，并用不同颜色的标注文本区分。
- 指标完善：
- 结余率边界处理（当月收入为0时显示为NaN并在hover中提示“收入为0，结余率不可计算”）。
- 引入预算列（如budget_dining、budget_transport），在图中叠加预算线，辅助超支识别。
- 配色与主题：
- 支持深色主题（dark）与高对比主题（colorblind-friendly），提升可访问性。
- 数据接口与验证：
- 增加数据校验函数，检查负值、缺失值与异常极值，提供清洗建议但不篡改原始数据。
- 支持多币种场景（自动换算到统一单位或分面显示）。
如需，我可以将以上模块封装为独立Python包，附带更详细的文档与单元测试模板，方便在您的个人理财应用中直接集成。
