"""
2.3 产品测试效率对比执行脚本
功能：按子产品维度分析测试周期、转化率、流失率等效能指标，并输出Excel、Markdown和图片
"""
import os
from datetime import datetime
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
# 设置中文显示
plt.rcParams["font.sans-serif"] = ["SimHei"]
plt.rcParams["axes.unicode_minus"] = False
# 配置
INPUT_FILE = "脱敏数据_特征增强.xlsx"
OUTPUT_DIR = "执行分析结果/2.3_产品测试效率对比"
OUTPUT_EXCEL = "2.3_产品测试效率对比.xlsx"
OUTPUT_MD = "2.3_产品测试效率对比摘要.md"
OUTPUT_PNG_RANK = "2.3_产品效能排名.png"
OUTPUT_PNG_COMPARE = "2.3_产品效能对比.png"
def main():
# 1. 读取数据
base_dir = os.path.dirname(os.path.dirname(__file__))
input_path = os.path.join(base_dir, INPUT_FILE)
if not os.path.exists(input_path):
raise FileNotFoundError(f"未找到输入文件：{input_path}")
df = pd.read_excel(input_path)
print(f"数据读取成功: {len(df)} 行, {len(df.columns)} 列")
# 创建输出目录
output_dir_abs = os.path.join(base_dir, OUTPUT_DIR)
os.makedirs(output_dir_abs, exist_ok=True)
excel_path = os.path.join(output_dir_abs, OUTPUT_EXCEL)
md_path = os.path.join(output_dir_abs, OUTPUT_MD)
png_rank_path = os.path.join(output_dir_abs, OUTPUT_PNG_RANK)
png_compare_path = os.path.join(output_dir_abs, OUTPUT_PNG_COMPARE)
# 必要字段检查
required_cols = [
"子产品名称",
"测试周期（天）",
"是否已接入",
"是否调用（数值）",
"是否不接入",
"不接入原因",
]
missing = [c for c in required_cols if c not in df.columns]
if missing:
raise ValueError(f"输入数据缺少必要字段: {missing}")
# =========================
# 1. 产品基础效能排名
# =========================
print("\n计算产品基础效能排名...")
# 记录总数
product_total = df.groupby("子产品名称").size().rename("记录总数")
# 测试周期统计
cycle_stats = df.groupby("子产品名称")["测试周期（天）"].agg(
平均测试周期="mean",
中位数测试周期="median",
标准差测试周期="std",
)
# 数量统计
agg_counts = df.groupby("子产品名称").agg(
已接入数量=("是否已接入", "sum"),
已调用数量=("是否调用（数值）", "sum"),
不接入数量=("是否不接入", "sum"),
)
product_stats = pd.concat([product_total, cycle_stats, agg_counts], axis=1)
# 计算转化率和流失率
product_stats["申请→已接入转化率(%)"] = (
np.where(
product_stats["记录总数"] > 0,
product_stats["已接入数量"] / product_stats["记录总数"] * 100,
np.nan,
)
).round(2)
product_stats["申请→已调用转化率(%)"] = (
np.where(
product_stats["记录总数"] > 0,
product_stats["已调用数量"] / product_stats["记录总数"] * 100,
np.nan,
)
).round(2)
product_stats["不接入流失率(%)"] = (
np.where(
product_stats["记录总数"] > 0,
product_stats["不接入数量"] / product_stats["记录总数"] * 100,
np.nan,
)
).round(2)
# 综合效能得分（按转化率排名）
valid_conv = product_stats["申请→已接入转化率(%)"].fillna(0)
product_stats["转化率得分"] = valid_conv.rank(pct=True) * 100
product_stats["综合效能得分"] = product_stats["转化率得分"].round(2)
product_stats["综合效能排名"] = (
product_stats["综合效能得分"].rank(ascending=False, method="min").astype(int)
)
product_stats = product_stats.sort_values("综合效能排名")
# =========================
# 2. 产品不接入原因分布
# =========================
print("计算产品不接入原因分布...")
if df["不接入原因"].notna().any():
tmp = df[df["不接入原因"].notna()]
product_reason = (
tmp.groupby("子产品名称")["不接入原因"]
.value_counts()
.unstack(fill_value=0)
)
reason_size = tmp.groupby("子产品名称").size()
product_reason_pct = (
product_reason.div(reason_size, axis=0).fillna(0) * 100
).round(2)
else:
product_reason = pd.DataFrame()
product_reason_pct = pd.DataFrame()
# =========================
# 3. 高效产品与低效产品
# =========================
print("识别高效产品与低效产品...")
stats_for_rank = product_stats.dropna(subset=["申请→已接入转化率(%)"]).copy()
if len(stats_for_rank) >= 2:
top_products = (
stats_for_rank.sort_values("申请→已接入转化率(%)", ascending=False)
.head(2)
.index.tolist()
)
bottom_products = (
stats_for_rank.sort_values("申请→已接入转化率(%)", ascending=True)
.head(2)
.index.tolist()
)
else:
top_products = stats_for_rank.index.tolist()
bottom_products = stats_for_rank.index.tolist()
def build_product_feature_dict(name: str) -> dict:
sub = df[df["子产品名称"] == name]
if name not in product_stats.index:
return {}
row = product_stats.loc[name]
return {
"记录总数": int(row["记录总数"]),
"转化率": float(row["申请→已接入转化率(%)"])
if pd.notna(row["申请→已接入转化率(%)"])
else np.nan,
"平均测试周期": float(row["平均测试周期"])
if pd.notna(row["平均测试周期"])
else np.nan,
"流失率": float(row["不接入流失率(%)"])
if pd.notna(row["不接入流失率(%)"])
else np.nan,
}
top_product_features = {
p: build_product_feature_dict(p) for p in top_products
}
bottom_product_features = {
p: build_product_feature_dict(p) for p in bottom_products
}
# =========================
# 4. 保存 Excel
# =========================
print(f"\n保存Excel文件: {excel_path}")
with pd.ExcelWriter(excel_path, engine="openpyxl") as writer:
product_stats.to_excel(writer, sheet_name="产品基础效能排名")
if not product_reason.empty:
product_reason.to_excel(writer, sheet_name="产品不接入原因分布")
product_reason_pct.to_excel(writer, sheet_name="产品不接入原因占比")
print("Excel文件保存成功")
# =========================
# 5. 生成图片
# =========================
print(f"\n生成图片...")
# 5.1 产品效能排名图
if not product_stats.empty:
products_for_plot = (
product_stats.sort_values("申请→已接入转化率(%)", ascending=False).head(
20
)
)
plt.figure(figsize=(12, 6))
plt.bar(
range(len(products_for_plot)),
products_for_plot["申请→已接入转化率(%)"],
color="steelblue",
)
plt.xticks(
range(len(products_for_plot)),
products_for_plot.index,
rotation=45,
ha="right",
)
plt.ylabel("申请→已接入转化率(%)")
plt.title("2.3 产品效能排名（按申请→已接入转化率）")
plt.tight_layout()
plt.savefig(png_rank_path, dpi=300, bbox_inches="tight")
plt.close()
print(f"产品效能排名图已保存: {png_rank_path}")
# 5.2 高效vs低效产品对比图
names_for_compare = list(dict.fromkeys(top_products + bottom_products))
if names_for_compare and len(names_for_compare) > 0:
metrics = [
"申请→已接入转化率(%)",
"平均测试周期",
"不接入流失率(%)",
]
comp_df = product_stats.loc[names_for_compare, metrics].copy()
x = np.arange(len(names_for_compare))
width = 0.25
plt.figure(figsize=(12, 6))
for i, m in enumerate(metrics):
plt.bar(
x + (i - 1) * width,
comp_df[m],
width=width,
label=m,
)
plt.xticks(x, [str(n) for n in names_for_compare], rotation=45, ha="right")
plt.ylabel("数值")
plt.title("2.3 高效 vs 低效产品对比（转化率 / 周期 / 流失率）")
plt.legend()
plt.tight_layout()
plt.savefig(png_compare_path, dpi=300, bbox_inches="tight")
plt.close()
print(f"产品效能对比图已保存: {png_compare_path}")
# =========================
# 6. 生成 Markdown 摘要
# =========================
print(f"\n生成Markdown摘要: {md_path}")
now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
md_lines = []
md_lines.append("# 产品测试效率对比摘要")
md_lines.append("")
md_lines.append(f"> **生成时间**: {now_str} ")
md_lines.append(f"> **数据文件**: {INPUT_FILE} ")
md_lines.append(f"> **数据量**: {len(df)} 条记录")
md_lines.append("")
md_lines.append("---")
md_lines.append("")
md_lines.append("## 📊 产品基础效能排名（按综合效能得分）")
md_lines.append("")
md_lines.append(
"| 排名 | 子产品名称 | 记录总数 | 平均测试周期(天) | 申请→已接入转化率(%) | "
"申请→已调用转化率(%) | 不接入流失率(%) | 综合效能得分 |"
)
md_lines.append(
"|------|------------|----------|------------------|---------------------|"
"---------------------|-----------------|--------------|"
)
for name, row in product_stats.iterrows():
md_lines.append(
f"| {int(row['综合效能排名'])} | {name} | "
f"{int(row['记录总数'])} | "
f"{row['平均测试周期']:.1f} | "
f"{row['申请→已接入转化率(%)']:.2f} | "
f"{row['申请→已调用转化率(%)']:.2f} | "
f"{row['不接入流失率(%)']:.2f} | "
f"{row['综合效能得分']:.2f} |"
)
# 高效产品
md_lines.append("")
md_lines.append("---")
md_lines.append("")
md_lines.append("## 🏆 高效产品特征（转化率 Top 2）")
md_lines.append("")
for p in top_products:
feat = top_product_features.get(p, {})
if not feat:
continue
md_lines.append(f"### {p}")
md_lines.append("")
md_lines.append(f"- **记录总数**: {feat['记录总数']}")
if not np.isnan(feat["转化率"]):
md_lines.append(f"- **申请→已接入转化率**: {feat['转化率']:.2f}%")
if not np.isnan(feat["平均测试周期"]):
md_lines.append(f"- **平均测试周期**: {feat['平均测试周期']:.1f} 天")
if not np.isnan(feat["流失率"]):
md_lines.append(f"- **不接入流失率**: {feat['流失率']:.2f}%")
md_lines.append("")
# 低效产品
md_lines.append("")
md_lines.append("---")
md_lines.append("")
md_lines.append("## ⚠️ 低效产品特征（转化率 Bottom 2）")
md_lines.append("")
for p in bottom_products:
feat = bottom_product_features.get(p, {})
if not feat:
continue
md_lines.append(f"### {p}")
md_lines.append("")
md_lines.append(f"- **记录总数**: {feat['记录总数']}")
if not np.isnan(feat["转化率"]):
md_lines.append(f"- **申请→已接入转化率**: {feat['转化率']:.2f}%")
if not np.isnan(feat["平均测试周期"]):
md_lines.append(f"- **平均测试周期**: {feat['平均测试周期']:.1f} 天")
if not np.isnan(feat["流失率"]):
md_lines.append(f"- **不接入流失率**: {feat['流失率']:.2f}%")
md_lines.append("")
# 关键发现
md_lines.append("")
md_lines.append("---")
md_lines.append("")
md_lines.append("## 📈 关键发现")
md_lines.append("")
if not product_stats.empty:
best = product_stats["申请→已接入转化率(%)"].idxmax()
worst = product_stats["申请→已接入转化率(%)"].idxmin()
md_lines.append(
f"1. **转化率最高产品**：{best} "
f"（{product_stats.loc[best, '申请→已接入转化率(%)']:.2f}%）"
)
md_lines.append(
f"2. **转化率最低产品**：{worst} "
f"（{product_stats.loc[worst, '申请→已接入转化率(%)']:.2f}%）"
)
md_lines.append(
f"3. **平均测试周期最短产品**："
f"{product_stats['平均测试周期'].idxmin()} "
f"（{product_stats['平均测试周期'].min():.1f} 天）"
)
md_lines.append(
f"4. **平均测试周期最长产品**："
f"{product_stats['平均测试周期'].idxmax()} "
f"（{product_stats['平均测试周期'].max():.1f} 天）"
)
md_lines.append(
f"5. **流失率最低产品**："
f"{product_stats['不接入流失率(%)'].idxmin()} "
f"（{product_stats['不接入流失率(%)'].min():.2f}%）"
)
md_lines.append(
f"6. **流失率最高产品**："
f"{product_stats['不接入流失率(%)'].idxmax()} "
f"（{product_stats['不接入流失率(%)'].max():.2f}%）"
)
md_lines.append("")
md_lines.append("---")
md_lines.append("")
md_lines.append("*本报告由 2.3_产品测试效率对比.md 方案自动生成*")
with open(md_path, "w", encoding="utf-8") as f:
f.write("\n".join(md_lines))
print("Markdown摘要生成成功")
print(f"\n✅ 产品测试效率对比分析完成！")
print(f" Excel文件: {excel_path}")
print(f" Markdown摘要: {md_path}")
print(f" 产品效能排名图: {png_rank_path}")
print(f" 产品效能对比图: {png_compare_path}")
if __name__ == "__main__":
main()