# -*- coding: utf-8 -*-
"""
============================================================
台湾上市公司财务破产风险分析 · 专业金融财务风可视化看板
============================================================
基于《基于台湾上市公司财务数据的破产分析.ipynb》(已修正/已运行)
与 output/models_comparison_notebook.csv、permutation_importance_notebook.csv
启动方式:
streamlit run streamlit_app.py
页面:
1. 决策总览 - 核心 KPI / 风险总览 / 模型推荐
2. 数据探索 - 数据结构、缺失、标识列画像
3. 特征风险画像 - 财务指标与破产风险的关联（EDA 结论互证）
4. 模型对比 - 11 个方案横向对比（Recall/F1/PR-AUC/ROC-AUC）
5. 特征重要性 - Permutation Importance + SHAP（如已生成）
6. 单公司体检 - 输入关键财务指标 → 破产概率（演示模型）
"""
import os
import numpy as np
import pandas as pd
import streamlit as st
import plotly.express as px
import plotly.graph_objects as go
# ============================================================
# 页面配置与金融财务风主题（深蓝 · 金色 · 衬线标题）
# ============================================================
st.set_page_config(
page_title="台湾上市公司财务破产风险分析看板",
page_icon="🏦",
layout="wide",
initial_sidebar_state="expanded",
)
FIN_CSS = """
"""
st.markdown(FIN_CSS, unsafe_allow_html=True)
def st_image_full_width(img_path):
"""兼容不同 streamlit 版本的图片宽度参数：
1.36 使用 use_column_width；新版（1.49+）改用 width='stretch'。"""
try:
st.image(img_path, use_column_width=True)
except TypeError:
try:
st.image(img_path, width="stretch")
except TypeError:
st.image(img_path)
# ============================================================
# 数据加载（缓存）
# ============================================================
BASE = os.path.dirname(os.path.abspath(__file__))
DATA_PATH = os.path.join(BASE, "data.csv")
OUT = os.path.join(BASE, "output")
@st.cache_data(show_spinner=False)
def load_data():
df = pd.read_csv(DATA_PATH)
df.columns = [i.title().strip() for i in list(df.columns)]
return df
@st.cache_data(show_spinner=False)
def load_model_csv():
p = os.path.join(OUT, "models_comparison_notebook.csv")
if os.path.exists(p):
return pd.read_csv(p)
return None
@st.cache_data(show_spinner=False)
def load_perm_csv():
p = os.path.join(OUT, "permutation_importance_notebook.csv")
if os.path.exists(p):
return pd.read_csv(p)
return None
data = load_data()
models_df = load_model_csv()
perm_df = load_perm_csv()
TARGET = "Bankrupt?"
int_cols = data.dtypes[data.dtypes == "int64"].index.tolist()
num_cols = [c for c in data.columns if c not in int_cols]
# 恒值列（建模时剔除）
constant_cols = [c for c in data.columns if data[c].nunique() <= 1]
def fmt_pct(x):
return f"{x*100:.2f}%"
# ============================================================
# 品牌区
# ============================================================
st.markdown(
f"""
台湾上市公司财务破产风险分析 · 决策看板
CORPORATE BANKRUPTCY EARLY-WARNING DASHBOARD · 财务金融 · 企业战略视角
样本 {data.shape[0]:,} 家 · 财务指标 {data.shape[1]-1} 项 · 破产样本 {int(data[TARGET].sum()):,} 家（{fmt_pct(data[TARGET].mean())}）· 模型方案 {0 if models_df is None else len(models_df)} 个
""",
unsafe_allow_html=True,
)
# ============================================================
# 侧边栏导航
# ============================================================
st.sidebar.markdown("## 🏦 破产风险分析")
st.sidebar.caption("台湾 1999–2009 上市公司财务数据")
page = st.sidebar.radio(
"导航",
["1. 决策总览", "2. 数据探索", "3. 特征风险画像",
"4. 模型对比", "5. 特征重要性", "6. 单公司体检"],
)
st.sidebar.markdown("---")
st.sidebar.caption("分析口径：Recall / F1 / PR-AUC 优先（类别不平衡）")
# ============================================================
# 页面 1：决策总览
# ============================================================
if page == "1. 决策总览":
st.markdown('核心指标总览
', unsafe_allow_html=True)
best = None
if models_df is not None and len(models_df) > 0:
best = models_df.sort_values("F1 score", ascending=False).iloc[0]
c1, c2, c3, c4, c5 = st.columns(5)
c1.markdown(
f'样本总量
{data.shape[0]:,}
'
f'
破产样本占比
{data[TARGET].mean()*100:.2f}%
'
f'
财务指标数量
{data.shape[1]-1}
'
f'
推荐模型 · F1
{best["F1 score"]:.3f}
'
f'
推荐模型 · PR-AUC
{best["PR-AUC score"]:.3f}
'
f'
破产分布与数据健康
', unsafe_allow_html=True)
colA, colB = st.columns([1, 1.1])
with colA:
dist = data[TARGET].value_counts().rename({0: "未破产 (0)", 1: "破产 (1)"})
fig = px.pie(
values=dist.values, names=dist.index,
title="目标变量分布（严重不平衡）",
color_discrete_sequence=["#0e7c7b", "#c9a227"],
hole=0.45,
)
fig.update_traces(textinfo="label+percent", textfont_size=13)
fig.update_layout(margin=dict(t=50, b=10, l=10, r=10))
st.plotly_chart(fig, use_container_width=True)
with colB:
health = pd.DataFrame({
"检查项": ["缺失值", "恒值列（建模剔除）", "数值型特征", "整数/标识列"],
"数量": [
int(data.isnull().sum().sum()),
len(constant_cols),
len(num_cols),
len(int_cols),
],
"说明": [
"无缺失，可直接建模",
"; ".join(constant_cols) if constant_cols else "无",
"标准化后进入模型",
"Bankrupt? / 标识列保持 0-1",
],
})
st.dataframe(health, use_container_width=True, hide_index=True)
st.markdown(
'⚠️ 数据无公司/年份标识：随机切分存在"伪独立"乐观偏差，'
'补充 CompanyId 后应改用 GroupKFold 复核（详见 notebook）。
',
unsafe_allow_html=True)
if best is not None:
st.markdown('战略结论速览
', unsafe_allow_html=True)
st.markdown(
f"""
推荐部署：{best["Algorithm"]}（Balance={best["Balance"]}，特征选择={best["FeatureSelection"]}）
—— F1={best["F1 score"]:.3f}，Recal={best["Recall"]:.3f}，PR-AUC={best["PR-AUC score"]:.3f}。
业务要点：① 96.8% 多数类下 accuracy 无意义，以 Recall/F1/PR-AUC 为准；
② 预期漏判率 ≈ {1-best["Recall"]:.1%}（漏检破产公司比例）；
③ 建议将本模型作为"财务预警红灯"前置环节，触发后人工复核。
""",
unsafe_allow_html=True)
# ============================================================
# 页面 2：数据探索
# ============================================================
elif page == "2. 数据探索":
st.markdown('数据结构与质量
', unsafe_allow_html=True)
c1, c2, c3 = st.columns(3)
c1.metric("样本行数", f"{data.shape[0]:,}")
c2.metric("特征列数", data.shape[1])
c3.metric("缺失值", int(data.isnull().sum().sum()))
st.markdown('数据类型分布
', unsafe_allow_html=True)
dtype_summary = (data.dtypes.astype(str).value_counts()
.rename_axis("数据类型").reset_index(name="列数"))
st.dataframe(dtype_summary, use_container_width=True, hide_index=True)
st.markdown('标识列画像（与破产的关系）
', unsafe_allow_html=True)
flag_cols = [c for c in int_cols if c != TARGET]
if flag_cols:
tabs = st.tabs(flag_cols)
for tab, fc in zip(tabs, flag_cols):
with tab:
cross = data.groupby([fc, TARGET]).size().reset_index(name="count")
cross["Bankrupt?"] = cross[TARGET].map({0: "未破产", 1: "破产"})
fig = px.bar(
cross, x=fc, y="count", color="Bankrupt?",
barmode="group",
title=f"{fc} × 破产交叉分布",
color_discrete_map={"未破产": "#0e7c7b", "破产": "#c9a227"},
text_auto=True,
)
st.plotly_chart(fig, use_container_width=True)
vc = data[fc].value_counts()
st.markdown(
f'{fc}：取值分布 {dict(vc)}。'
f'{"⚠️ 该列在建模时被剔除（恒值，无区分度）。" if fc in constant_cols else "该列保留建模。"}
',
unsafe_allow_html=True)
else:
st.info("无其他标识列。")
st.markdown('描述统计（前 12 个财务指标）
', unsafe_allow_html=True)
st.dataframe(data[num_cols].describe().T.head(12), use_container_width=True)
# ============================================================
# 页面 3：特征风险画像
# ============================================================
elif page == "3. 特征风险画像":
st.markdown('财务指标与破产风险的线性关联（Pearson）
', unsafe_allow_html=True)
corr = data[num_cols].corrwith(data[TARGET]).sort_values()
top_neg = corr.head(8)
top_pos = corr.tail(8).sort_values(ascending=False)
df_corr = pd.DataFrame({
"指标": list(top_pos.index) + list(top_neg.index),
"与破产相关性": list(top_pos.values) + list(top_neg.values),
})
colors = ["#b23a48" if v > 0 else "#0e7c7b" for v in df_corr["与破产相关性"]]
fig = px.bar(
df_corr, x="与破产相关性", y="指标", orientation="h",
title="Top 16 财务指标 × Bankrupt? 相关性（正=高风险 · 负=保护性）",
color="与破产相关性", color_continuous_scale=["#0e7c7b", "#e6e9ef", "#b23a48"],
)
fig.update_layout(yaxis=dict(autorange="reversed"), height=520)
st.plotly_chart(fig, use_container_width=True)
st.markdown(
'📌 与 EDA 互证：高负债类指标（债务比率、流动负债/资产等）与破产正相关；'
'资产厚、ROA 高、现金流充沛类指标呈保护性。模型重要性应与此互证（见「特征重要性」页）。
',
unsafe_allow_html=True)
st.markdown('关键指标：破产 vs 未破产 分布对比
', unsafe_allow_html=True)
key_feats = [
"Debt Ratio %", "Current Liability To Assets", "Current Liability To Current Assets",
"Roa(A) Before Interest And % After Tax", "Cash Flow To Total Assets",
"Interest Coverage Ratio (Interest Expense To Ebit)",
]
key_feats = [k for k in key_feats if k in data.columns][:6]
if key_feats:
cols = st.columns(3)
for i, k in enumerate(key_feats):
with cols[i % 3]:
tmp = data[[k, TARGET]].copy()
tmp["group"] = tmp[TARGET].map({0: "未破产", 1: "破产"})
fig = px.box(
tmp, x="group", y=k, color="group",
color_discrete_map={"未破产": "#0e7c7b", "破产": "#c9a227"},
title=k, points=False,
)
fig.update_layout(showlegend=False, height=260, margin=dict(t=40))
st.plotly_chart(fig, use_container_width=True)
# ============================================================
# 页面 4：模型对比
# ============================================================
elif page == "4. 模型对比":
st.markdown('模型方案横向对比（Recall / F1 / PR-AUC 优先）
', unsafe_allow_html=True)
if models_df is None:
st.error("未找到 output/models_comparison_notebook.csv，请先在 notebook 中运行模型对比。")
else:
disp = models_df.copy()
disp["Model Score"] = disp["Model Score"].astype(str)
show_cols = ["Algorithm", "FeatureSelection", "Balance", "Model Score",
"Precision", "Recall", "F1 score", "ROC-AUC score", "PR-AUC score"]
st.dataframe(disp[show_cols].sort_values("F1 score", ascending=False),
use_container_width=True, hide_index=True)
st.markdown('Recall / F1 / ROC-AUC 对比
', unsafe_allow_html=True)
metric = st.radio("选择指标", ["F1 score", "Recall", "ROC-AUC score", "PR-AUC score"],
horizontal=True, index=0)
sdf = disp.sort_values(metric, ascending=True)
fig = px.bar(
sdf, x=metric, y="Algorithm", orientation="h",
color="FeatureSelection",
color_discrete_map={"No": "#0e7c7b", "Yes": "#c9a227"},
title=f"{metric} by Model（无特征选择 vs 有特征选择）",
text=metric,
)
fig.update_layout(height=max(420, 30 * len(sdf)), yaxis=dict(autorange="reversed"))
st.plotly_chart(fig, use_container_width=True)
best = disp.sort_values("F1 score", ascending=False).iloc[0]
st.markdown(
f"""
业务推荐：{best["Algorithm"]}（Balance={best["Balance"]}，特征选择={best["FeatureSelection"]}）
F1={best["F1 score"]:.3f} · Recall={best["Recall"]:.3f} · PR-AUC={best["PR-AUC score"]:.3f} · ROC-AUC={best["ROC-AUC score"]:.3f}
最优超参：{best.get("BestParams", "—")}
决策口径：漏判率 = 1 − Recall ≈ {1-best["Recall"]:.1%}；
建议 SMOTE 与 class_weight 双轨验证后，选择漏判更低且 F1 不显著下降者。
""",
unsafe_allow_html=True)
# ============================================================
# 页面 5：特征重要性
# ============================================================
elif page == "5. 特征重要性":
st.markdown('特征重要性 · Permutation Importance（最优模型）
', unsafe_allow_html=True)
if perm_df is None:
st.warning("未找到 output/permutation_importance_notebook.csv。")
else:
topn = st.slider("展示前 N 个特征", 5, min(30, len(perm_df)), 15)
ptop = perm_df.head(topn).iloc[::-1]
fig = px.bar(
ptop, x="importance", y="feature", orientation="h",
title=f"Top {topn} Permutation Importance（scoring=f1）",
color="importance", color_continuous_scale="Viridis",
text="importance",
)
fig.update_layout(height=max(420, 26 * topn))
st.plotly_chart(fig, use_container_width=True)
st.markdown('战略解读
', unsafe_allow_html=True)
top3 = perm_df.head(3)["feature"].tolist()
st.markdown(
f"""
Top 驱动因子：{'、'.join(top3)}。
战略含义：① 高负债依赖（Borrowing Dependency、Current Liability To Assets）是破产的首要先行信号，
应纳入贷后/投资尽调的强制红灯指标；② ROA、现金流类指标排名靠前说明盈利质量与现金造血能力
对存续至关重要；③ 建议将重要性 Top 指标组合为"财务健康评分卡"，按月滚动监控阈值变化。
""",
unsafe_allow_html=True)
shap_top_path = os.path.join(OUT, "shap_top_notebook.csv")
if os.path.exists(shap_top_path):
st.markdown('SHAP Top 特征（平均 |SHAP|）
', unsafe_allow_html=True)
try:
shap_top = pd.read_csv(shap_top_path, index_col=0)
shap_top.columns = ["mean|SHAP|"]
shap_top = shap_top.reset_index().rename(columns={"index": "feature"})
st.dataframe(shap_top.head(15), use_container_width=True, hide_index=True)
except Exception:
pass
shap_img = os.path.join(OUT, "shap_summary_notebook.png")
if os.path.exists(shap_img):
st.markdown('SHAP 摘要图
', unsafe_allow_html=True)
st_image_full_width(shap_img)
else:
st.caption("（notebook 中 SHAP 为尽力而为，未生成图时以 permutation importance 为准）")
# ============================================================
# 页面 6：单公司体检（交互预测）
# ============================================================
elif page == "6. 单公司体检":
st.markdown('单公司破产风险快速体检（演示模型）
', unsafe_allow_html=True)
st.markdown(
'输入关键财务指标（标准分/原始值），基于 Random Forest + SMOTE 演示模型给出破产概率。'
'该页仅作分析演示，正式使用请以 notebook 训练集口径校准阈值。
',
unsafe_allow_html=True)
@st.cache_resource(show_spinner=True)
def train_demo_model():
from imblearn.over_sampling import SMOTE
from sklearn.ensemble import RandomForestClassifier
d = data.copy()
d = d.drop(columns=[c for c in d.columns if d[c].nunique() <= 1])
numf = d.dtypes[d.dtypes != "int64"].index
means = d[numf].mean()
stds = d[numf].std()
d[numf] = d[numf].apply(lambda x: (x - x.mean()) / (x.std()))
d[numf] = d[numf].fillna(0)
X = d.drop(TARGET, axis=1)
y = d[TARGET]
sm = SMOTE(sampling_strategy="minority", random_state=42)
Xr, yr = sm.fit_resample(X, y)
model = RandomForestClassifier(n_estimators=300, max_depth=None,
random_state=42, n_jobs=-1)
model.fit(Xr, yr)
return model, X.columns.tolist(), means, stds
with st.spinner("训练演示模型（首次约 10-30 秒）..."):
model, feats, means, stds = train_demo_model()
input_feats = [
"Debt Ratio %", "Current Liability To Assets", "Current Liability To Current Assets",
"Borrowing Dependency", "Roa(A) Before Interest And % After Tax",
"Net Income To Total Assets", "Cash Flow To Total Assets",
"Interest Coverage Ratio (Interest Expense To Ebit)",
"Working Capital To Total Assets", "Quick Assets/Current Liability",
]
input_feats = [f for f in input_feats if f in feats][:10]
st.markdown('输入财务指标（原始值）
', unsafe_allow_html=True)
cols = st.columns(2)
vals = {}
for i, f in enumerate(input_feats):
with cols[i % 2]:
med = float(data[f].median())
vals[f] = st.number_input(
f, value=round(med, 4), step=0.01,
help=f"中位数参考: {med:.4f} | 均值: {data[f].mean():.4f}",
)
if st.button("🚨 评估破产风险", type="primary", use_container_width=True):
row = pd.Series(0.0, index=feats, dtype=float)
for f in input_feats:
if f in means.index:
row[f] = (vals[f] - means[f]) / stds[f] if stds[f] else 0.0
proba = model.predict_proba(row.values.reshape(1, -1))[0, 1]
risk = "低风险" if proba < 0.2 else ("中等风险" if proba < 0.5 else "高风险")
c1, c2 = st.columns([1, 1.2])
with c1:
fig = go.Figure(go.Indicator(
mode="gauge+number",
value=round(proba * 100, 2),
number={"suffix": "%"},
title={"text": "破产概率"},
gauge={
"axis": {"range": [0, 100]},
"bar": {"color": "#b23a48" if proba >= 0.5 else ("#e0b84c" if proba >= 0.2 else "#0e7c7b")},
"steps": [
{"range": [0, 20], "color": "#e7f4ee"},
{"range": [20, 50], "color": "#fdf3e3"},
{"range": [50, 100], "color": "#fbe9ec"},
],
},
))
fig.update_layout(height=300)
st.plotly_chart(fig, use_container_width=True)
with c2:
cls = {"低风险": "risk-low", "中等风险": "risk-mid", "高风险": "risk-high"}[risk]
st.markdown(
f'评估结论
'
f'{risk}：破产概率约 {proba*100:.1f}%。'
f'{"建议立即触发人工财务复核与预警。" if proba >= 0.5 else ("建议纳入常规监控。" if proba >= 0.2 else "财务状况稳健，按常规跟踪。")}
',
unsafe_allow_html=True)
st.caption("注：演示模型基于全量样本 SMOTE 重训，概率阈值 0.2/0.5 为演示设定，请按业务容忍度校准。")
st.sidebar.markdown("---")
st.sidebar.caption("数据来源：data.csv · 模型结果：output/models_comparison_notebook.csv")