|
|
@@ -1,1679 +0,0 @@
|
|
|
-"""
|
|
|
-Deep-analysis insight generators for weekly and monthly reports.
|
|
|
-Each function returns list[dict] with {'title': str, 'content': str}.
|
|
|
-"""
|
|
|
-
|
|
|
-
|
|
|
-# =============================================================================
|
|
|
-# WEEKLY INSIGHTS
|
|
|
-# =============================================================================
|
|
|
-
|
|
|
-def weekly_insights(page_type: str, metrics: dict, context: dict) -> list[dict]:
|
|
|
- """Dispatch to specific weekly page insight functions."""
|
|
|
- dispatch = {
|
|
|
- 'weekly_summary': _insight_weekly_summary,
|
|
|
- 'weekly_trend': _insight_weekly_trend,
|
|
|
- 'weekly_wow': _insight_weekly_wow,
|
|
|
- 'weekly_region': _insight_weekly_region,
|
|
|
- 'weekly_country': _insight_weekly_country,
|
|
|
- 'weekly_team': _insight_weekly_team,
|
|
|
- 'weekly_issue': _insight_weekly_issue,
|
|
|
- 'weekly_plan': _insight_weekly_plan,
|
|
|
- }
|
|
|
- fn = dispatch.get(page_type)
|
|
|
- return fn(metrics, context) if fn else []
|
|
|
-
|
|
|
-
|
|
|
-def _insight_weekly_summary(metrics: dict, context: dict) -> list[dict]:
|
|
|
- """Page 2: 周内节奏分析、月度进度、关键驱动因素"""
|
|
|
- items = []
|
|
|
- daily_trend = metrics.get('daily_trend', {})
|
|
|
- total_qty = metrics.get('total_qty', 0)
|
|
|
- tracking_orders = metrics.get('tracking_orders', 0)
|
|
|
- prev_tracking_orders = metrics.get('prev_tracking_orders', 0)
|
|
|
- top_countries = metrics.get('top_countries', {})
|
|
|
- region_dist = metrics.get('region_dist', {})
|
|
|
-
|
|
|
- # ① 周内节奏分析
|
|
|
- if daily_trend:
|
|
|
- dates = list(daily_trend.keys())
|
|
|
- vals = list(daily_trend.values())
|
|
|
- peak_idx = max(range(len(vals)), key=lambda i: vals[i])
|
|
|
- low_idx = min(range(len(vals)), key=lambda i: vals[i])
|
|
|
- peak_date, peak_val = dates[peak_idx], vals[peak_idx]
|
|
|
- low_date, low_val = dates[low_idx], vals[low_idx]
|
|
|
- prev_avg = metrics.get('prev_avg_daily_orders', 0)
|
|
|
- above_avg = sum(1 for v in vals if prev_avg and v > prev_avg)
|
|
|
- rhythm = (
|
|
|
- f'本周订单呈"{peak_date}冲高、{low_date}回落"特征，峰值{peak_val}单、低谷{low_val}单，'
|
|
|
- f'波动幅度{_safe_div(peak_val, low_val) if low_val else 0:.1f}倍。'
|
|
|
- f'{"高于" if above_avg >= len(vals)//2 else "低于"}上周均值天数占比{above_avg}/{len(vals)}，'
|
|
|
- f'整体节奏{"前移" if peak_idx <= len(vals)//2 else "后移"}。'
|
|
|
- f'建议分析低谷日是否受节假日或客户账期影响，优化排产与沟通节奏。'
|
|
|
- )
|
|
|
- else:
|
|
|
- rhythm = (
|
|
|
- f'本周累计订单{tracking_orders}单、{total_qty}台，'
|
|
|
- f'日均{metrics.get("avg_daily_orders", 0)}单。'
|
|
|
- f'由于缺乏分日数据，暂无法识别周内节奏特征。'
|
|
|
- f'建议完善日报 granularity，以便识别周几效应并优化资源投放节奏。'
|
|
|
- )
|
|
|
- items.append({'title': '📅 周内节奏分析', 'content': rhythm})
|
|
|
-
|
|
|
- # ② 与上周对比偏移
|
|
|
- wow_pct = _pct_change(tracking_orders, prev_tracking_orders)
|
|
|
- if prev_tracking_orders:
|
|
|
- shift = (
|
|
|
- f'本周{tracking_orders}单较上周{prev_tracking_orders}单{_fmt_chg_dir(wow_pct)}{_fmt_pct(wow_pct)}，'
|
|
|
- f'{"呈加速态势" if (wow_pct or 0) > 10 else "增长放缓" if (wow_pct or 0) > 0 else "出现回落"}。'
|
|
|
- f'若趋势持续，下月累计将{"超额" if (wow_pct or 0) > 0 else "缺口"}约{abs((wow_pct or 0)):.0f}%月度目标。'
|
|
|
- f'建议结合pipeline深度评估：增长来自存量转化还是新增签约，两种来源的可持续性差异显著。'
|
|
|
- )
|
|
|
- else:
|
|
|
- shift = (
|
|
|
- f'本周订单量{tracking_orders}单，缺乏上周同期数据作为基准。'
|
|
|
- f'建议尽快建立周环比追踪机制，识别趋势方向。当前可依据日均{metrics.get("avg_daily_orders", 0)}单推算月度节奏，'
|
|
|
- f'若保持稳定，整月预计约{metrics.get("avg_daily_orders", 0) * 30:.0f}单。'
|
|
|
- )
|
|
|
- items.append({'title': '📈 周环比趋势偏移', 'content': shift})
|
|
|
-
|
|
|
- # ③ 月度进度推演
|
|
|
- monthly_target = context.get('monthly_target', 0)
|
|
|
- if monthly_target and total_qty:
|
|
|
- progress_pct = round(total_qty / monthly_target * 100, 1)
|
|
|
- week_of_month = context.get('week_of_month', 1)
|
|
|
- expected = week_of_month / 4 * 100
|
|
|
- gap = progress_pct - expected
|
|
|
- progress = (
|
|
|
- f'本周完成{total_qty}台，占月度目标{monthly_target}台的{progress_pct}%，'
|
|
|
- f'按"四周均衡"预期应达{expected:.1f}%，{"领先" if gap >= 0 else "落后"}{abs(gap):.1f}个百分点。'
|
|
|
- f'{"若保持当前 pace，月度有望超额完成。" if gap >= 0 else "剩余周需加速追赶，建议启动冲刺机制。"}'
|
|
|
- f'重点关注最后一周产能与舱位是否匹配冲刺需求。'
|
|
|
- )
|
|
|
- else:
|
|
|
- progress = (
|
|
|
- f'本周完成{total_qty}台，因未设定月度目标或缺乏台数数据，暂无法计算进度占比。'
|
|
|
- f'建议业务部门明确月度销量目标，以便周报自动生成目标达成率并触发预警。'
|
|
|
- f'当前可先以本周为基准，要求后续每周环比增长不低于5%作为软约束。'
|
|
|
- )
|
|
|
- items.append({'title': '🎯 月度进度推演', 'content': progress})
|
|
|
-
|
|
|
- # ④ 关键驱动因素（国家维度）
|
|
|
- if top_countries:
|
|
|
- top1_country = list(top_countries.keys())[0]
|
|
|
- first_val = list(top_countries.values())[0]
|
|
|
- top1_qty = first_val if isinstance(first_val, int) else first_val.get('qty', 0)
|
|
|
- top3_qty = 0
|
|
|
- for v in list(top_countries.values())[:3]:
|
|
|
- top3_qty += v if isinstance(v, int) else v.get('qty', 0)
|
|
|
- concentration = round(top3_qty / total_qty * 100, 1) if total_qty else 0
|
|
|
- driver = (
|
|
|
- f'本周Top 3国家贡献{top3_qty}台（占比{concentration}%），其中{top1_country}以{top1_qty}台领跑。'
|
|
|
- f'{"大客户驱动特征明显，单国依赖度较高" if concentration > 40 else "国家分布相对均衡，抗波动能力较强"}。'
|
|
|
- f'若Top 1国家订单来自单一客户，建议评估该客户下月复购概率；'
|
|
|
- f'若来自多客户分散订单，则说明该国市场渗透进入良性循环。'
|
|
|
- )
|
|
|
- else:
|
|
|
- driver = (
|
|
|
- f'本周累计{total_qty}台，暂无国家维度分解数据。'
|
|
|
- f'建议完善订单数据中国家字段，以便识别关键驱动市场并制定差异化资源投入策略。'
|
|
|
- f'当前阶段可通过负责人访谈定性判断：本周增长主要来自老客户返单还是新市场突破。'
|
|
|
- )
|
|
|
- items.append({'title': '💡 关键驱动因素', 'content': driver})
|
|
|
-
|
|
|
- # ⑤ 区域引擎识别
|
|
|
- if region_dist:
|
|
|
- regions = sorted(region_dist.items(), key=lambda x: x[1].get('qty', 0), reverse=True)
|
|
|
- r1_name, r1_data = regions[0]
|
|
|
- r1_qty = r1_data.get('qty', 0)
|
|
|
- r1_pct = r1_data.get('pct', 0)
|
|
|
- top_c = [c["country"] for c in r1_data.get("top_countries", [])[:2]]
|
|
|
- region_text = (
|
|
|
- f'{r1_name}本周贡献{r1_qty}台（{r1_pct}%），为当前第一大区域引擎。'
|
|
|
- f'其Top国家为{"、".join(top_c)}。'
|
|
|
- f'建议对该区域实施"深耕+复制"策略：在成熟国家推进增值服务提升客单价，'
|
|
|
- f'同时将成功经验复制至同区域二线国家，最大化区域协同效应。'
|
|
|
- )
|
|
|
- else:
|
|
|
- region_text = (
|
|
|
- f'本周订单{tracking_orders}单，缺乏区域维度数据。'
|
|
|
- f'建议建立国家→区域映射表，以便自动识别区域引擎并评估资源投入ROI。'
|
|
|
- f'当前可通过订单号前缀或客户归属地手动标注区域，作为过渡方案。'
|
|
|
- )
|
|
|
- items.append({'title': '🌍 区域引擎识别', 'content': region_text})
|
|
|
-
|
|
|
- # ⑥ 结构健康度速览
|
|
|
- avg_qty = metrics.get('avg_qty_per_order', 0)
|
|
|
- prev_avg_qty = metrics.get('prev_avg_qty', 0) or context.get('prev_avg_qty', 0)
|
|
|
- avg_chg = _pct_change(avg_qty, prev_avg_qty)
|
|
|
- health = (
|
|
|
- f'本周单均台数{avg_qty}台，{"较上周" + _fmt_chg_dir(avg_chg) + _fmt_pct(avg_chg) if prev_avg_qty else "基准待建立"}。'
|
|
|
- f'{"大单占比提升，客户结构优化" if (avg_chg or 0) > 0 else "中小单为主，需关注大客户转化" if avg_qty else "数据缺失"}。'
|
|
|
- f'结合{metrics.get("countries", 0)}个目的国分析，'
|
|
|
- f'{"市场覆盖广但单国深度不足" if metrics.get("countries", 0) > 15 and avg_qty < 20 else ""}'
|
|
|
- f'建议下周重点跟进单均台数大于50台的大单，锁定其对月度目标的贡献。'
|
|
|
- )
|
|
|
- items.append({'title': '⚖️ 结构健康度速览', 'content': health})
|
|
|
-
|
|
|
- return items
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-def _insight_weekly_trend(metrics: dict, context: dict) -> list[dict]:
|
|
|
- """Page 3: 周内波动归因、与上周结构性差异、趋势持续性"""
|
|
|
- items = []
|
|
|
- daily_trend = metrics.get('daily_trend', {})
|
|
|
- prev_daily_trend = context.get('prev_daily_trend', {})
|
|
|
- tracking_orders = metrics.get('tracking_orders', 0)
|
|
|
- prev_tracking_orders = metrics.get('prev_tracking_orders', 0)
|
|
|
- forecast_next = metrics.get('forecast_next', 0)
|
|
|
- total_qty = metrics.get('total_qty', 0)
|
|
|
-
|
|
|
- # ① 周内波动归因
|
|
|
- if daily_trend:
|
|
|
- dates = list(daily_trend.keys())
|
|
|
- vals = list(daily_trend.values())
|
|
|
- avg_val = sum(vals) / len(vals)
|
|
|
- low_days = [d for d, v in daily_trend.items() if v < avg_val * 0.7]
|
|
|
- high_days = [d for d, v in daily_trend.items() if v > avg_val * 1.3]
|
|
|
- fluctuation = (
|
|
|
- f'本周日均{avg_val:.0f}单，{"、".join(high_days) if high_days else "无"}为显著高于均值的高活跃日，'
|
|
|
- f'{"、".join(low_days) if low_days else "无"}为低于均值0.7倍的低谷日。'
|
|
|
- f'低谷日需排查是否为节假日（当地法定假期）、客户月度/季度账期结算日，或系统批量录入延迟。'
|
|
|
- f'建议建立"低谷日归因登记簿"，记录每次低谷的外部因素，积累预测模型训练数据。'
|
|
|
- )
|
|
|
- else:
|
|
|
- fluctuation = (
|
|
|
- f'本周累计{tracking_orders}单，缺乏分日趋势数据。'
|
|
|
- f'建议每日统计新增订单数，以便识别周内波动模式。当前可通过负责人日报定性判断：'
|
|
|
- f'本周是否有1-2天明显低于其他工作日，若有，需排查外部干扰因素并提前制定应对措施。'
|
|
|
- )
|
|
|
- items.append({'title': '🔍 周内波动归因', 'content': fluctuation})
|
|
|
-
|
|
|
- # ② 与上周结构性差异
|
|
|
- if daily_trend and prev_daily_trend:
|
|
|
- prev_vals = list(prev_daily_trend.values())
|
|
|
- curr_vals = list(daily_trend.values())
|
|
|
- curr_shape = "前高后低" if sum(curr_vals[:len(curr_vals)//2]) > sum(curr_vals[len(curr_vals)//2:]) else "前低后高"
|
|
|
- prev_shape = "前高后低" if sum(prev_vals[:len(prev_vals)//2]) > sum(prev_vals[len(prev_vals)//2:]) else "前低后高"
|
|
|
- structural = (
|
|
|
- f'本周走势形态为"{curr_shape}"，上周为"{prev_shape}"，{"形态一致" if curr_shape == prev_shape else "形态发生反转"}。'
|
|
|
- f'{"说明客户下单节奏稳定，周末效应或账期规律持续生效。" if curr_shape == prev_shape else "说明外部条件发生变化，可能受节假日错位或大单时点影响。"}'
|
|
|
- f'建议对比两周的Top 3客户下单日期，判断结构差异来自系统性因素还是偶然大单。'
|
|
|
- )
|
|
|
- else:
|
|
|
- structural = (
|
|
|
- f'本周订单{tracking_orders}单，'
|
|
|
- f'{"较上周" + _fmt_chg_dir(_pct_change(tracking_orders, prev_tracking_orders)) + _fmt_pct(_pct_change(tracking_orders, prev_tracking_orders)) if prev_tracking_orders else "缺乏上周分日数据"}。'
|
|
|
- f'建议保留每周分日数据，以便进行走势形态对比（前高后低 vs 前低后高）。'
|
|
|
- f'形态一致性是判断趋势可持续性的重要信号。'
|
|
|
- )
|
|
|
- items.append({'title': '📊 与上周结构性差异', 'content': structural})
|
|
|
-
|
|
|
- # ③ 趋势持续性判断
|
|
|
- wow_pct = _pct_change(tracking_orders, prev_tracking_orders)
|
|
|
- if forecast_next:
|
|
|
- sustain = (
|
|
|
- f'Pipeline预测下月交付{forecast_next}台，结合本周{tracking_orders}单（wow {_fmt_pct(wow_pct)}），'
|
|
|
- f'{"趋势向上且pipeline充足，持续性较强" if (wow_pct or 0) > 5 and forecast_next > total_qty * 0.8 else ""}'
|
|
|
- f'{"趋势向好但pipeline不足，需警惕后继乏力" if (wow_pct or 0) > 5 and forecast_next <= total_qty * 0.5 else ""}'
|
|
|
- f'{"趋势承压，pipeline也偏保守，双重承压" if (wow_pct or 0) < 0 else ""}'
|
|
|
- f'。建议基于A+B阶段存量订单数评估真实转化潜力，'
|
|
|
- f'若早期pipeline充足，则本周增长具备延续性；若依赖存量冲刺，则下周可能回落。'
|
|
|
- )
|
|
|
- else:
|
|
|
- sustain = (
|
|
|
- f'本周订单{tracking_orders}单（wow {_fmt_pct(wow_pct)}），因缺乏pipeline预测数据，'
|
|
|
- f'趋势持续性需通过阶段结构间接判断。'
|
|
|
- f'若A+B阶段订单占比大于40%，说明前期储备充足，下周有望维持；'
|
|
|
- f'若E+F阶段占比过高，则本周增长可能来自集中交付的一次性脉冲，持续性存疑。'
|
|
|
- )
|
|
|
- items.append({'title': '🔮 趋势持续性判断', 'content': sustain})
|
|
|
-
|
|
|
- # ④ 外部因素关联
|
|
|
- support_cats = metrics.get('support_categories', {})
|
|
|
- if support_cats:
|
|
|
- top_cat = max(support_cats.items(), key=lambda x: x[1])
|
|
|
- external = (
|
|
|
- f'本周支持需求中"{top_cat[0]}"类最多（{top_cat[1]}项），'
|
|
|
- f'{"多为物流/船期问题，说明交付端承压，可能影响客户下单信心" if "物流" in top_cat[0] or "船期" in top_cat[0] else ""}'
|
|
|
- f'{"多为财务/收款问题，说明资金端存在卡点，可能拖慢合同锁定节奏" if "财务" in top_cat[0] or "收款" in top_cat[0] else ""}'
|
|
|
- f'。建议将该类问题的根因归类为流程/政策/外部三类，'
|
|
|
- f'流程类内部优化，政策类提前预警，外部类建立替代方案（如备用船期/汇率对冲）。'
|
|
|
- )
|
|
|
- else:
|
|
|
- external = (
|
|
|
- f'本周暂无支持需求数据。建议主动收集：汇率波动、目的国进口政策调整、船期延误等外部事件，'
|
|
|
- f'并评估其对本周订单节奏的干扰程度。'
|
|
|
- f'建立"外部因素-订单波动"关联看板，可显著提升趋势归因的准确性。'
|
|
|
- )
|
|
|
- items.append({'title': '🌐 外部因素关联', 'content': external})
|
|
|
-
|
|
|
- # ⑤ 异常点识别
|
|
|
- if daily_trend:
|
|
|
- vals = list(daily_trend.values())
|
|
|
- mean_v = sum(vals) / len(vals)
|
|
|
- std = (sum((v - mean_v) ** 2 for v in vals) / len(vals)) ** 0.5
|
|
|
- outliers = [(d, v) for d, v in daily_trend.items() if abs(v - mean_v) > 1.5 * std]
|
|
|
- if outliers:
|
|
|
- outlier_text = (
|
|
|
- f'本周{"、".join([d for d, _ in outliers])}出现异常波动，偏离均值{mean_v:.0f}单超过1.5倍标准差。'
|
|
|
- f'异常日需区分"机会型"（大客户集中下单）与"风险型"（系统故障/数据补录）。'
|
|
|
- f'建议对异常日逐单标注原因，长期积累后可建立异常检测规则，实现自动化预警。'
|
|
|
- )
|
|
|
- else:
|
|
|
- outlier_text = (
|
|
|
- f'本周各日订单围绕均值{mean_v:.0f}单正常波动，无显著异常点。'
|
|
|
- f'说明业务运行平稳，外部干扰较少。建议保持当前运营节奏，'
|
|
|
- f'同时将本周作为"平稳基线"保存，用于未来异常检测的对比基准。'
|
|
|
- )
|
|
|
- else:
|
|
|
- outlier_text = (
|
|
|
- f'缺乏分日数据，无法识别周内异常点。'
|
|
|
- f'建议建立日报机制后，采用"均值+1.5倍标准差"规则自动标记异常交易日。'
|
|
|
- )
|
|
|
- items.append({'title': '⚠️ 异常点识别', 'content': outlier_text})
|
|
|
-
|
|
|
- # ⑥ 下周走势预判
|
|
|
- if daily_trend and prev_daily_trend:
|
|
|
- curr_avg = sum(daily_trend.values()) / len(daily_trend)
|
|
|
- prev_avg = sum(prev_daily_trend.values()) / len(prev_daily_trend)
|
|
|
- momentum = _pct_change(curr_avg, prev_avg)
|
|
|
- pred_low = int(curr_avg * 0.85)
|
|
|
- pred_high = int(curr_avg * 1.15)
|
|
|
- forecast = (
|
|
|
- f'基于本周日均{curr_avg:.0f}单及环比动量{_fmt_pct(momentum)}，'
|
|
|
- f'预测下周日均订单区间{pred_low}-{pred_high}单，整周预计{pred_low * 5}-{pred_high * 5}单。'
|
|
|
- f'{"若大客户持续下单，有望突破区间上限" if (momentum or 0) > 10 else ""}'
|
|
|
- f'{"若外部因素未改善，可能回落至区间下限" if (momentum or 0) < 0 else ""}'
|
|
|
- f'。建议每日晨会对照预测区间，偏差大于20%时触发专项复盘。'
|
|
|
- )
|
|
|
- else:
|
|
|
- forecast = (
|
|
|
- f'当前在跟订单{tracking_orders}单，基于历史经验，下周预计维持在'
|
|
|
- f'{int(tracking_orders * 0.85)}-{int(tracking_orders * 1.15)}单区间。'
|
|
|
- f'建议重点关注A→B阶段转化速率，这是下周增量的最可预测来源。'
|
|
|
- )
|
|
|
- items.append({'title': '📉 下周走势预判', 'content': forecast})
|
|
|
-
|
|
|
- return items
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-def _insight_weekly_wow(metrics: dict, context: dict) -> list[dict]:
|
|
|
- """Page 4: 各环节转化效率、瓶颈环节深度诊断、库存/资金占用"""
|
|
|
- items = []
|
|
|
- status_wow = metrics.get('status_wow', {})
|
|
|
- status_dist = metrics.get('status_dist', {})
|
|
|
- total_orders = metrics.get('tracking_orders', 0)
|
|
|
-
|
|
|
- def _get_status(name):
|
|
|
- return status_dist.get(name, 0)
|
|
|
-
|
|
|
- # ① 各环节转化效率
|
|
|
- a_name, b_name = '合同拟定中', '已锁定合同待付订金'
|
|
|
- a_curr = status_wow.get(a_name, {}).get('current', 0)
|
|
|
- a_prev = status_wow.get(a_name, {}).get('previous', 0)
|
|
|
- b_curr = status_wow.get(b_name, {}).get('current', 0)
|
|
|
- b_prev = status_wow.get(b_name, {}).get('previous', 0)
|
|
|
- if a_prev and b_prev:
|
|
|
- prev_conv = round(b_prev / a_prev * 100, 1)
|
|
|
- curr_conv = round(b_curr / a_curr * 100, 1) if a_curr else 0
|
|
|
- conv_chg = _pct_change(curr_conv, prev_conv)
|
|
|
- conv = (
|
|
|
- f'A→B阶段转化率本周{curr_conv}%（{b_curr}/{a_curr}），上周{prev_conv}%（{b_prev}/{a_prev}），'
|
|
|
- f'{_fmt_chg_dir(conv_chg)}{_fmt_pct(conv_chg)}。'
|
|
|
- f'{"转化率提升，说明合同推进效率改善" if (conv_chg or 0) > 0 else "转化率下降，合同锁定遇阻"}。'
|
|
|
- f'建议拆解转化失败的根因：客户侧（审批慢/资金未到位）vs 我方侧（条款争议/响应慢），针对性优化。'
|
|
|
- )
|
|
|
- else:
|
|
|
- conv = (
|
|
|
- f'本周A阶段{a_curr}单、B阶段{b_curr}单，因缺乏上周A/B分阶段数据，无法计算转化率变化。'
|
|
|
- f'建议建立阶段快照机制，每周固定时点统计各阶段存量。当前可先设定A→B周转化目标为A阶段存量的15%，'
|
|
|
- f'并落实到具体负责人。'
|
|
|
- )
|
|
|
- items.append({'title': '⚡ A→B转化效率', 'content': conv})
|
|
|
-
|
|
|
- # ② 瓶颈环节深度诊断
|
|
|
- changes = [(name, data.get('change_pct', 0) or 0) for name, data in status_wow.items()]
|
|
|
- if changes:
|
|
|
- worst = min(changes, key=lambda x: x[1])
|
|
|
- best = max(changes, key=lambda x: x[1])
|
|
|
- bottleneck = (
|
|
|
- f'本周降幅最大环节为"{worst[0]}"（{_fmt_pct(worst[1])}），增幅最大为"{best[0]}"（+{_fmt_pct(best[1])}）。'
|
|
|
- f'{"若降幅环节为E/F，说明交付端受阻；若为A/B，则前端获客或转化遇困。"}'
|
|
|
- f'建议对{worst[0]}环节启动"五问法"根因分析：'
|
|
|
- f'是流程卡点、资源不足、还是外部政策突变？找到根因后48h内出具改进方案。'
|
|
|
- )
|
|
|
- else:
|
|
|
- bottleneck = (
|
|
|
- f'本周各阶段环比数据不完整，暂无法识别瓶颈环节。'
|
|
|
- f'建议建立每周阶段快照对比机制，重点监控C→D（生产）和D→E（尾款）两个关键转化节点。'
|
|
|
- f'这两个节点直接决定交付周期和资金回笼速度，是 Weekly Review 的核心指标。'
|
|
|
- )
|
|
|
- items.append({'title': '📉 瓶颈环节深度诊断', 'content': bottleneck})
|
|
|
-
|
|
|
- # ③ 库存与资金占用分析
|
|
|
- c_curr = status_wow.get('已付订金待生产', {}).get('current', 0)
|
|
|
- d_curr = status_wow.get('已生产待付尾款', {}).get('current', 0)
|
|
|
- c_prev = status_wow.get('已付订金待生产', {}).get('previous', 0)
|
|
|
- d_prev = status_wow.get('已生产待付尾款', {}).get('previous', 0)
|
|
|
- cd_total = c_curr + d_curr
|
|
|
- cd_prev = c_prev + d_prev
|
|
|
- cd_chg = _pct_change(cd_total, cd_prev)
|
|
|
- inventory = (
|
|
|
- f'生产端（C+D）本周合计{cd_total}单，较上周{_fmt_chg_dir(cd_chg)}{_fmt_pct(cd_chg)}。'
|
|
|
- f'{"生产端积压加重，资金与产能占用上升" if (cd_chg or 0) > 10 else "生产端平稳，库存压力可控" if abs(cd_chg or 0) <= 10 else "生产端去化加速，周转改善"}。'
|
|
|
- f'D阶段{d_curr}单为核心风险点：已生产未收款，每单约占用产能和资金。'
|
|
|
- f'建议对D阶段大于14天的订单启动专项催收，缩短资金占用周期。'
|
|
|
- )
|
|
|
- items.append({'title': '💰 库存与资金占用分析', 'content': inventory})
|
|
|
-
|
|
|
- # ④ 发运端效率
|
|
|
- e_curr = status_wow.get('已付尾款待发运', {}).get('current', 0)
|
|
|
- f_curr = status_wow.get('已发运', {}).get('current', 0)
|
|
|
- e_prev = status_wow.get('已付尾款待发运', {}).get('previous', 0)
|
|
|
- ef_total = e_curr + f_curr
|
|
|
- ef_text = (
|
|
|
- f'发运端（E+F）本周合计{ef_total}单，其中待发运{e_curr}单、已发运{f_curr}单。'
|
|
|
- f'待发运较上周{_fmt_chg_dir(_pct_change(e_curr, e_prev))}{_fmt_pct(_pct_change(e_curr, e_prev))}。'
|
|
|
- f'{"待发运积压上升，需排查船期/舱位/报关瓶颈" if e_curr > (e_prev * 1.2 if e_prev else 0) else "发运节奏顺畅，交付闭环效率良好"}。'
|
|
|
- f'建议物流部门每周提供船期表，销售据此与客户对齐收货预期，减少催单消耗的管理精力。'
|
|
|
- )
|
|
|
- items.append({'title': '🚢 发运端效率', 'content': ef_text})
|
|
|
-
|
|
|
- # ⑤ 漏斗健康度综合评分
|
|
|
- if total_orders:
|
|
|
- early = _get_status('合同拟定中') + _get_status('已锁定合同待付订金')
|
|
|
- mid = _get_status('已付订金待生产') + _get_status('已生产待付尾款')
|
|
|
- late = _get_status('已付尾款待发运') + _get_status('已发运')
|
|
|
- early_pct = round(early / total_orders * 100, 1)
|
|
|
- mid_pct = round(mid / total_orders * 100, 1)
|
|
|
- late_pct = round(late / total_orders * 100, 1)
|
|
|
- health = (
|
|
|
- f'本周漏斗结构：前期{early_pct}%（{early}单）、中期{mid_pct}%（{mid}单）、后期{late_pct}%（{late}单）。'
|
|
|
- f'{"前期占比偏高，pipeline充足但转化压力较大" if early_pct > 40 else ""}'
|
|
|
- f'{"中期占比偏高，生产端承压，需关注产能瓶颈" if mid_pct > 40 else ""}'
|
|
|
- f'{"后期占比偏高，交付冲刺期，需确保物流资源到位" if late_pct > 40 else ""}'
|
|
|
- f'理想结构为前期35%-中期35%-后期30%，当前{"接近理想" if 30 <= early_pct <= 40 and 30 <= mid_pct <= 40 else "偏离理想，需针对性调整"}。'
|
|
|
- )
|
|
|
- else:
|
|
|
- health = (
|
|
|
- f'本周暂无订单数据，无法计算漏斗结构。'
|
|
|
- f'建议建立标准化漏斗健康度评分模型：前期35%-中期35%-后期30%为基准，'
|
|
|
- f'偏离超过10个百分点时自动触发预警并推荐改进动作。'
|
|
|
- )
|
|
|
- items.append({'title': '🏥 漏斗健康度综合评分', 'content': health})
|
|
|
-
|
|
|
- return items
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-def _insight_weekly_region(metrics: dict, context: dict) -> list[dict]:
|
|
|
- """Page 5: 区域战略优先级、区域间协同、新兴市场孵化"""
|
|
|
- items = []
|
|
|
- region_dist = metrics.get('region_dist', {})
|
|
|
- total_qty = metrics.get('total_qty', 0)
|
|
|
- prev_region_dist = context.get('prev_region_dist', {})
|
|
|
-
|
|
|
- if not region_dist:
|
|
|
- return [
|
|
|
- {'title': '💡 区域战略优先级', 'content': '本周暂无区域分布数据。建议完善订单数据中的国家与区域映射，以便识别明星区域与问题区域，优化资源投放策略。'},
|
|
|
- {'title': '📈 区域间协同', 'content': '缺乏区域维度数据，暂无法评估区域间协同效应。建议收集相邻区域订单关联数据，识别是否存在客户转介绍或区域联动带来的增长。'},
|
|
|
- ]
|
|
|
-
|
|
|
- # ① 区域战略优先级矩阵
|
|
|
- regions = []
|
|
|
- for name, data in region_dist.items():
|
|
|
- qty = data.get('qty', 0)
|
|
|
- pct = data.get('pct', 0)
|
|
|
- prev_qty = prev_region_dist.get(name, {}).get('qty', 0) if prev_region_dist else 0
|
|
|
- growth = _pct_change(qty, prev_qty)
|
|
|
- regions.append((name, qty, pct, growth))
|
|
|
-
|
|
|
- stars = [r for r in regions if r[2] > 20 and (r[3] is None or r[3] > 0)]
|
|
|
- cows = [r for r in regions if r[2] > 20 and (r[3] is not None and r[3] <= 0)]
|
|
|
- questions = [r for r in regions if r[2] <= 20 and (r[3] is None or r[3] > 0)]
|
|
|
- dogs = [r for r in regions if r[2] <= 20 and (r[3] is not None and r[3] <= 0)]
|
|
|
-
|
|
|
- matrix = (
|
|
|
- f'基于"占比×增速"矩阵分析：'
|
|
|
- f'{"明星区域（高占比+增长）：" + "、".join([r[0] for r in stars[:2]]) + "，应继续加大投入；" if stars else ""}'
|
|
|
- f'{"现金牛（高占比+放缓）：" + "、".join([r[0] for r in cows[:2]]) + "，维持投入收割利润；" if cows else ""}'
|
|
|
- f'{"问题区域（低占比+增长）：" + "、".join([r[0] for r in questions[:2]]) + "，需评估是否加大培育；" if questions else ""}'
|
|
|
- f'{"瘦狗区域（低占比+下滑）：" + "、".join([r[0] for r in dogs[:2]]) + "，考虑收缩或调整策略。" if dogs else ""}'
|
|
|
- )
|
|
|
- if not any([stars, cows, questions, dogs]):
|
|
|
- matrix = (
|
|
|
- f'本周各区域占比与增速数据不足以完成四象限分类。'
|
|
|
- f'建议建立区域级周环比追踪，当区域数据完整后可自动生成波士顿矩阵并推荐资源配置策略。'
|
|
|
- )
|
|
|
- items.append({'title': '💡 区域战略优先级矩阵', 'content': matrix})
|
|
|
-
|
|
|
- # ② 区域间协同效应
|
|
|
- top_regions = sorted(regions, key=lambda x: x[1], reverse=True)[:3]
|
|
|
- if len(top_regions) >= 2:
|
|
|
- synergy = (
|
|
|
- f'{top_regions[0][0]}本周贡献{top_regions[0][1]}台领跑，{top_regions[1][0]}以{top_regions[1][1]}台紧随其后。'
|
|
|
- f'若两区域存在客户转介绍或同一代理商覆盖，则具备协同放大效应。'
|
|
|
- f'建议复盘{top_regions[0][0]}的成功经验（车型偏好/渠道模式/定价策略），'
|
|
|
- f'形成标准化打法后向{top_regions[1][0]}及同类区域复制，降低试错成本。'
|
|
|
- )
|
|
|
- else:
|
|
|
- synergy = (
|
|
|
- f'本周{top_regions[0][0]}为绝对主导区域，其他区域占比偏低。'
|
|
|
- f'单极结构下协同效应有限，建议评估是否通过"成熟区带新区"机制，'
|
|
|
- f'让成熟区域负责人兼任相邻新兴市场顾问，实现经验外溢。'
|
|
|
- )
|
|
|
- items.append({'title': '📈 区域间协同效应', 'content': synergy})
|
|
|
-
|
|
|
- # ③ 新兴市场孵化
|
|
|
- top8_countries = set()
|
|
|
- for data in region_dist.values():
|
|
|
- for c in data.get('top_countries', []):
|
|
|
- top8_countries.add(c.get('country', ''))
|
|
|
- all_countries_count = metrics.get('countries', 0)
|
|
|
- if all_countries_count > len(top8_countries):
|
|
|
- emerging = (
|
|
|
- f'本周覆盖{all_countries_count}国，Top 8框架内国家{len(top8_countries)}个，'
|
|
|
- f'另有{all_countries_count - len(top8_countries)}国为框架外新兴市场。'
|
|
|
- f'若框架外国家本周出现首单或复购，说明市场孵化取得突破。'
|
|
|
- f'建议对框架外国家建立"观察名单"，单月订单大于3台即纳入正式跟踪，并匹配专项资源。'
|
|
|
- )
|
|
|
- else:
|
|
|
- emerging = (
|
|
|
- f'本周覆盖{all_countries_count}国，订单高度集中在Top国家。'
|
|
|
- f'新兴市场孵化进度较慢，建议设定"每月突破1个新国家"的拓展目标，'
|
|
|
- f'通过参加区域性车展、与当地经销商建立合作等方式扩大市场覆盖。'
|
|
|
- )
|
|
|
- items.append({'title': '🌱 新兴市场孵化', 'content': emerging})
|
|
|
-
|
|
|
- # ④ 区域投入ROI评估
|
|
|
- if regions:
|
|
|
- growths = [r[3] for r in regions if r[3] is not None]
|
|
|
- avg_growth = sum(growths) / len(growths) if growths else 0
|
|
|
- roi = (
|
|
|
- f'本周区域平均增速{_fmt_pct(avg_growth)}。'
|
|
|
- f'{"增速为正的区域的投入产出比优于整体，建议将增量预算向这些区域倾斜" if avg_growth > 0 else "整体增速承压，建议收缩低效区域投入，集中资源保高潜市场"}。'
|
|
|
- f'ROI评估应综合考虑订单量、利润率和售后成本：高订单低利润区域需谨慎追加投入，'
|
|
|
- f'低订单高利润区域（如配件服务收入高）反而值得深耕。'
|
|
|
- )
|
|
|
- else:
|
|
|
- roi = (
|
|
|
- f'缺乏区域增速数据，暂无法计算投入ROI。'
|
|
|
- f'建议建立区域级损益表，至少包含订单量、毛利率、物流成本、售后成本四项指标，'
|
|
|
- f'每季度评估一次区域真实贡献度，优化资源分配。'
|
|
|
- )
|
|
|
- items.append({'title': '💰 区域投入ROI评估', 'content': roi})
|
|
|
-
|
|
|
- # ⑤ 区域风险分散
|
|
|
- top1_region_pct = max(r[2] for r in regions) if regions else 0
|
|
|
- risk = (
|
|
|
- f'第一大区域占比{top1_region_pct}%，{"集中度偏高，存在单区域政策/汇率波动风险" if top1_region_pct > 50 else "集中度适中，风险分散良好"}。'
|
|
|
- f'{"建议3个月内将第一大区域占比压降至50%以下，通过培育第二梯队实现结构优化" if top1_region_pct > 50 else "建议继续巩固现有均衡格局，同时培育1-2个潜力区域作为第三极"}。'
|
|
|
- f'区域多元化是抵御单一市场政策风险的最有效手段。'
|
|
|
- )
|
|
|
- items.append({'title': '⚠️ 区域风险分散', 'content': risk})
|
|
|
-
|
|
|
- return items
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-def _insight_weekly_country(metrics: dict, context: dict) -> list[dict]:
|
|
|
- """Page 6: 国家组合健康度、大客户集中度、竞争格局"""
|
|
|
- items = []
|
|
|
- top_countries = metrics.get('top_countries', {})
|
|
|
- top_countries_change = metrics.get('top_countries_change', {})
|
|
|
- total_qty = metrics.get('total_qty', 0)
|
|
|
- top6_concentration_pct = metrics.get('top6_concentration_pct', 0)
|
|
|
-
|
|
|
- if not top_countries:
|
|
|
- return [
|
|
|
- {'title': '💡 国家组合健康度', 'content': '本周暂无国家分布数据。建议完善目的国家字段，以便评估国家组合集中度风险并制定分散策略。'},
|
|
|
- ]
|
|
|
-
|
|
|
- country_qty = {}
|
|
|
- for c, v in top_countries.items():
|
|
|
- country_qty[c] = v if isinstance(v, int) else v.get('qty', 0)
|
|
|
-
|
|
|
- sorted_countries = sorted(country_qty.items(), key=lambda x: x[1], reverse=True)
|
|
|
- top3_qty = sum(v for _, v in sorted_countries[:3])
|
|
|
- top3_pct = round(top3_qty / total_qty * 100, 1) if total_qty else 0
|
|
|
-
|
|
|
- # ① 国家组合健康度
|
|
|
- health = (
|
|
|
- f'Top 3国家合计{top3_qty}台（{top3_pct}%），{"超过40%警戒线，单国波动对整体业绩影响显著" if top3_pct > 40 else "低于40%，国家分布相对健康"}。'
|
|
|
- f'{"Top 6集中度" + str(top6_concentration_pct) + "%进一步印证了大客户/大国家驱动特征" if top6_concentration_pct > 60 else ""}'
|
|
|
- f'建议设定"Top 3占比小于40%、Top 6小于65%"的组合健康目标，'
|
|
|
- f'通过培育第二梯队国家（4-8名）逐步降低头部依赖。'
|
|
|
- )
|
|
|
- items.append({'title': '💡 国家组合健康度', 'content': health})
|
|
|
-
|
|
|
- # ② 大客户集中度
|
|
|
- top1_country, top1_qty = sorted_countries[0]
|
|
|
- chg_data = top_countries_change.get(top1_country, {})
|
|
|
- chg_pct = chg_data.get('change_pct') if isinstance(chg_data, dict) else None
|
|
|
- concentration = (
|
|
|
- f'{top1_country}本周{top1_qty}台领跑，环比{_fmt_pct(chg_pct) if chg_pct is not None else "—"}。'
|
|
|
- f'若该国订单来自单一客户，则存在极大客户依赖风险，该客户流失将导致月度目标大幅缺口。'
|
|
|
- f'建议对Top 1国家进行客户拆解：单一客户占比大于50%则触发红色预警，需立即启动新客户开发计划。'
|
|
|
- )
|
|
|
- items.append({'title': '👤 大客户集中度风险', 'content': concentration})
|
|
|
-
|
|
|
- # ③ 国家增速梯队
|
|
|
- if top_countries_change:
|
|
|
- growing = [(c, d['change_pct']) for c, d in top_countries_change.items() if d.get('change_pct', 0) and d['change_pct'] > 0]
|
|
|
- declining = [(c, d['change_pct']) for c, d in top_countries_change.items() if d.get('change_pct', 0) and d['change_pct'] < 0]
|
|
|
- growing.sort(key=lambda x: x[1], reverse=True)
|
|
|
- declining.sort(key=lambda x: x[1])
|
|
|
- tier = (
|
|
|
- f'国家增速梯队：上升最快{"为" + "、".join([c for c, _ in growing[:2]]) if growing else "暂无"}，'
|
|
|
- f'下滑最快{"为" + "、".join([c for c, _ in declining[:2]]) if declining else "暂无"}。'
|
|
|
- f'上升国家需分析是政策红利还是客户拓展见效，判断可持续性；'
|
|
|
- f'下滑国家需区分暂时性因素（节假日/账期）还是结构性因素（竞争加剧/需求萎缩），对症下药。'
|
|
|
- )
|
|
|
- else:
|
|
|
- tier = (
|
|
|
- f'本周缺乏国家维度环比数据，无法划分增速梯队。'
|
|
|
- f'建议每周固定输出Top 10国家环比变化表，识别"黑马"与"掉队"国家，'
|
|
|
- f'为资源动态调配提供数据依据。'
|
|
|
- )
|
|
|
- items.append({'title': '📊 国家增速梯队', 'content': tier})
|
|
|
-
|
|
|
- # ④ 竞争格局判断
|
|
|
- competition = (
|
|
|
- f'若{top1_country}市场出现价格竞争或交付周期压缩，说明竞品正在渗透。'
|
|
|
- f'建议建立"竞争信号"监测机制：客户询盘时提及竞品频次、询盘转化率变化、订单周期拉长等。'
|
|
|
- f'本周若Top 1国家订单增速放缓但询盘量上升，可能是竞品截流信号，需立即调整报价或服务策略。'
|
|
|
- )
|
|
|
- items.append({'title': '🏁 竞争格局判断', 'content': competition})
|
|
|
-
|
|
|
- # ⑤ 国家生命周期策略
|
|
|
- if len(sorted_countries) >= 2:
|
|
|
- mid = sorted_countries[len(sorted_countries)//2][1] if sorted_countries else 0
|
|
|
- mature = [c for c, v in sorted_countries if v >= mid * 1.2]
|
|
|
- emerging = [c for c, v in sorted_countries if v < mid * 0.8]
|
|
|
- lifecycle = (
|
|
|
- f'成熟市场（{"、".join(mature[:3])}）订单稳定，建议主推高毛利车型和金融服务提升客单价；'
|
|
|
- f'新兴市场（{"、".join(emerging[:3])}）处于首单突破期，当前应以保交付口碑为核心，'
|
|
|
- f'建立标杆案例后通过客户转介绍实现低成本获客。两种市场的KPI应差异化设置。'
|
|
|
- )
|
|
|
- else:
|
|
|
- lifecycle = (
|
|
|
- f'当前国家数量不足以划分生命周期梯队。'
|
|
|
- f'建议积累更多国家数据后，按"订单量+复购率"双维度将国家分为导入期、成长期、成熟期、衰退期，'
|
|
|
- f'匹配差异化的产品、定价和服务策略。'
|
|
|
- )
|
|
|
- items.append({'title': '📈 国家生命周期策略', 'content': lifecycle})
|
|
|
-
|
|
|
- return items
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-def _insight_weekly_team(metrics: dict, context: dict) -> list[dict]:
|
|
|
- """Page 7: 人均产出趋势、区域专注度、协作效率"""
|
|
|
- items = []
|
|
|
- team = metrics.get('team', {})
|
|
|
- team_wow = metrics.get('team_wow', {})
|
|
|
- per_capita_orders = metrics.get('per_capita_orders', 0)
|
|
|
- prev_per_capita = context.get('prev_per_capita_orders', 0)
|
|
|
- countries = metrics.get('countries', 0)
|
|
|
-
|
|
|
- if isinstance(team, dict) and 'owners' in team:
|
|
|
- owners = team['owners']
|
|
|
- qty_map = team.get('qty', {})
|
|
|
- else:
|
|
|
- owners = {k: v.get('orders', 0) for k, v in team.items()} if team else {}
|
|
|
- qty_map = {k: v.get('qty', 0) for k, v in team.items()} if team else {}
|
|
|
-
|
|
|
- if not owners:
|
|
|
- return [
|
|
|
- {'title': '💡 人均产出趋势', 'content': '本周暂无负责人数据。建议完善订单归属字段，以便评估团队人效并优化负载分配。'},
|
|
|
- {'title': '⚖️ 区域专注度', 'content': '缺乏团队数据，无法评估区域专注度。建议建立人均覆盖国家数指标，超过8个时触发精力分散预警。'},
|
|
|
- ]
|
|
|
-
|
|
|
- n_members = len(owners)
|
|
|
- total_orders = sum(owners.values())
|
|
|
- sorted_owners = sorted(owners.items(), key=lambda x: x[1], reverse=True)
|
|
|
- top_owner, top_val = sorted_owners[0]
|
|
|
-
|
|
|
- # ① 人均产出趋势
|
|
|
- pc_chg = _pct_change(per_capita_orders, prev_per_capita)
|
|
|
- productivity = (
|
|
|
- f'本周团队{n_members}人，人均{per_capita_orders:.1f}单，'
|
|
|
- f'{"较上周" + _fmt_chg_dir(pc_chg) + _fmt_pct(pc_chg) if prev_per_capita else "基准待建立"}。'
|
|
|
- f'{"人均产出提升，团队效率改善" if (pc_chg or 0) > 0 else "人均产出下滑，需排查是人员增加稀释还是订单总量下降"}。'
|
|
|
- f'建议将人均产出纳入周会Review，连续两周下滑时启动专项诊断。'
|
|
|
- )
|
|
|
- items.append({'title': '💡 人均产出趋势', 'content': productivity})
|
|
|
-
|
|
|
- # ② 头部与尾部差距
|
|
|
- tail_vals = [v for _, v in sorted_owners[-3:]]
|
|
|
- tail_avg = sum(tail_vals) / len(tail_vals) if tail_vals else 0
|
|
|
- gap = _safe_div(top_val, tail_avg) if tail_avg else 0
|
|
|
- disparity = (
|
|
|
- f'团队头部{top_owner}本周{top_val}单，尾部3人均约{tail_avg:.1f}单，'
|
|
|
- f'头尾差距{gap:.1f}倍，{"差距较大，存在激励或能力分化" if gap > 3 else "差距适中，团队相对均衡"}。'
|
|
|
- f'若差距来自能力差异，建议安排头部带教；若来自资源分配不均（如国家/客户分配），'
|
|
|
- f'建议重新盘点客户池，向低负载负责人倾斜高潜客户。'
|
|
|
- )
|
|
|
- items.append({'title': '📊 头部与尾部差距', 'content': disparity})
|
|
|
-
|
|
|
- # ③ 区域专注度
|
|
|
- if countries and n_members:
|
|
|
- avg_countries = countries / n_members
|
|
|
- focus = (
|
|
|
- f'团队人均覆盖{avg_countries:.1f}个国家，'
|
|
|
- f'{"超过8个，精力高度分散，单国深度不足" if avg_countries > 8 else "在合理范围内，兼顾广度与深度"}。'
|
|
|
- f'{"建议将国家按优先级分为A/B/C类，A类由专人深耕，B类共享覆盖，C类交给代理商" if avg_countries > 8 else ""}'
|
|
|
- f'{"建议对重点国家实施双人backup机制，避免单点依赖" if avg_countries <= 5 else ""}'
|
|
|
- f'。国家覆盖数与订单量呈倒U型关系，过多或过少均不利。'
|
|
|
- )
|
|
|
- else:
|
|
|
- focus = (
|
|
|
- f'缺乏国家数或团队人数数据，无法计算人均覆盖。'
|
|
|
- f'建议建立"人均覆盖国家数"指标，警戒线为8个，超过时触发区域重新划分流程。'
|
|
|
- )
|
|
|
- items.append({'title': '⚖️ 区域专注度', 'content': focus})
|
|
|
-
|
|
|
- # ④ 协作效率
|
|
|
- cross_orders = context.get('cross_owner_orders', 0)
|
|
|
- if cross_orders:
|
|
|
- ratio = round(cross_orders / total_orders * 100, 1) if total_orders else 0
|
|
|
- collaboration = (
|
|
|
- f'本周跨负责人协作订单{cross_orders}单，占总量{ratio}%。'
|
|
|
- f'{"协作占比偏低，团队呈单兵作战状态，知识共享不足" if ratio < 5 else "协作机制运行良好，团队具备协同作战能力"}。'
|
|
|
- f'建议对大型项目（单均大于50台）强制要求双负责人制，既分散风险又促进经验交流。'
|
|
|
- )
|
|
|
- else:
|
|
|
- collaboration = (
|
|
|
- f'本周暂无跨负责人协作订单数据。建议建立协作订单标记机制，'
|
|
|
- f'识别需要多区域/多客户类型协同的复杂项目。'
|
|
|
- f'协作效率是团队从"单兵"向"集团军"转型的关键指标，应纳入季度考核。'
|
|
|
- )
|
|
|
- items.append({'title': '🤝 协作效率', 'content': collaboration})
|
|
|
-
|
|
|
- # ⑤ 人员稳定性
|
|
|
- if team_wow:
|
|
|
- churn = [o for o, d in team_wow.items() if d.get('previous', 0) > 0 and d.get('current', 0) == 0]
|
|
|
- new_rise = [o for o, d in team_wow.items() if d.get('previous', 0) == 0 and d.get('current', 0) > 0]
|
|
|
- stability = (
|
|
|
- f'团队变动：{"上周有产出但本周挂零：" + "、".join(churn[:2]) if churn else "无流失风险信号"}；'
|
|
|
- f'{"本周新涌现：" + "、".join(new_rise[:2]) if new_rise else "无新人冒头"}。'
|
|
|
- f'{"需关注挂零负责人的客户状态，是否离职/调岗/休假导致订单断档，必要时启动客户交接保护机制" if churn else "团队人员稳定，无异常流失或新增涌现，建议保持当前激励机制并关注长期职业发展路径"}'
|
|
|
- f'{"新涌现负责人值得复盘其成功经验，快速复制至团队" if new_rise else ""}'
|
|
|
- )
|
|
|
- else:
|
|
|
- stability = (
|
|
|
- f'缺乏上周团队对比数据，无法评估人员稳定性。'
|
|
|
- f'建议建立人员周环比追踪，连续两周挂零或大幅波动时触发主管一对一沟通。'
|
|
|
- f'人员稳定性是业务连续性的基础，突然变动往往导致客户流失和知识断层。'
|
|
|
- )
|
|
|
- items.append({'title': '👥 人员稳定性', 'content': stability})
|
|
|
-
|
|
|
- return items
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-def _insight_weekly_issue(metrics: dict, context: dict) -> list[dict]:
|
|
|
- """Page 8: 问题根因分类、影响面量化、重复性问题、解决进度"""
|
|
|
- items = []
|
|
|
- issues = metrics.get('issues', [])
|
|
|
- support_categories = metrics.get('support_categories', {})
|
|
|
- tracking_orders = metrics.get('tracking_orders', 0)
|
|
|
- total_qty = metrics.get('total_qty', 0)
|
|
|
-
|
|
|
- # ① 问题根因分类
|
|
|
- if issues:
|
|
|
- root_causes = {'技术': 0, '流程': 0, '外部': 0}
|
|
|
- for issue in issues:
|
|
|
- detail = issue.get('detail', '') + issue.get('title', '')
|
|
|
- if any(k in detail for k in ['系统', 'IT', '技术', '质量', '检测']):
|
|
|
- root_causes['技术'] += 1
|
|
|
- elif any(k in detail for k in ['流程', '审批', '协调', '内部']):
|
|
|
- root_causes['流程'] += 1
|
|
|
- else:
|
|
|
- root_causes['外部'] += 1
|
|
|
- dominant = max(root_causes.items(), key=lambda x: x[1])
|
|
|
- root = (
|
|
|
- f'本周{len(issues)}项问题按根因分类：技术类{root_causes["技术"]}项、流程类{root_causes["流程"]}项、外部类{root_causes["外部"]}项。'
|
|
|
- f'{"技术类最多，说明产品质量或系统稳定性存在短板，需产研部门介入" if dominant[0] == "技术" else ""}'
|
|
|
- f'{"流程类最多，说明内部协作存在断点，建议梳理跨部门SOP" if dominant[0] == "流程" else ""}'
|
|
|
- f'{"外部类最多，说明问题主要来自客户/政策/物流等不可控因素，需建立预案库" if dominant[0] == "外部" else ""}'
|
|
|
- f'。根因分类是预防性管理的基础，建议每次问题上报时强制选择根因类型。'
|
|
|
- )
|
|
|
- else:
|
|
|
- root = (
|
|
|
- f'本周暂无显式问题记录。建议建立"无问题也是一种信号"的视角：'
|
|
|
- f'若支持需求总量上升但问题记录为零，可能是问题未被识别或归类。'
|
|
|
- f'建议每周五由负责人提交本周卡点，即使已自行解决也记录备案，积累组织知识。'
|
|
|
- )
|
|
|
- items.append({'title': '🔍 问题根因分类', 'content': root})
|
|
|
-
|
|
|
- # ② 影响面量化
|
|
|
- if issues:
|
|
|
- est_orders = max(len(issues), sum(1 for i in issues if i.get('severity') == '高'))
|
|
|
- est_qty = est_orders * 30
|
|
|
- est_amount = est_orders * 150
|
|
|
- impact = (
|
|
|
- f'本周问题预计影响订单{est_orders}单、车辆约{est_qty}台、金额约¥{est_amount}万。'
|
|
|
- f'其中高严重度问题{"占比高，需立即升级至管理层" if est_orders > 3 else "可控，按常规流程处理即可"}。'
|
|
|
- f'建议建立"问题影响面"必填字段：阻断单数、预计台数、预计金额，便于后续按损失金额排序处理优先级。'
|
|
|
- )
|
|
|
- else:
|
|
|
- impact = (
|
|
|
- f'本周无显式问题，预计影响面为零。建议将节省的管理精力转向'
|
|
|
- f'"支持需求前置化处理"，把潜在问题消灭在萌芽阶段。'
|
|
|
- f'支持需求中若出现高频关键词，应视为早期预警信号。'
|
|
|
- )
|
|
|
- items.append({'title': '💰 影响面量化', 'content': impact})
|
|
|
-
|
|
|
- # ③ 重复性问题识别
|
|
|
- if support_categories:
|
|
|
- top_cat = max(support_categories.items(), key=lambda x: x[1])
|
|
|
- recurring = (
|
|
|
- f'本周支持需求中"{top_cat[0]}"类出现{top_cat[1]}次，为最高频类别。'
|
|
|
- f'若该类别连续两周及以上位居榜首，则可判定为重复性问题。'
|
|
|
- f'建议对{top_cat[0]}类问题启动"根治计划"：统计发生场景→提炼标准化处理模板→培训一线人员→设置系统校验规则，'
|
|
|
- f'目标是将该类问题发生频率压降50%以上。'
|
|
|
- )
|
|
|
- else:
|
|
|
- recurring = (
|
|
|
- f'本周无支持需求分类数据，无法识别重复性问题。'
|
|
|
- f'建议积累4周以上数据后，按"月度出现频次大于3次且连续两周上榜"定义重复性问题，'
|
|
|
- f'并建立专项改进小组负责根治。'
|
|
|
- )
|
|
|
- items.append({'title': '🔄 重复性问题识别', 'content': recurring})
|
|
|
-
|
|
|
- # ④ 上周问题解决率
|
|
|
- prev_resolved = context.get('prev_week_resolved_issues', 0)
|
|
|
- prev_total = context.get('prev_week_total_issues', 0)
|
|
|
- if prev_total:
|
|
|
- resolve_rate = round(prev_resolved / prev_total * 100, 1)
|
|
|
- resolution = (
|
|
|
- f'上周问题{prev_total}项，已解决{prev_resolved}项，解决率{resolve_rate}%。'
|
|
|
- f'{"解决率大于80%，问题闭环效率高" if resolve_rate > 80 else "解决率小于80%，存在遗留问题堆积风险"}。'
|
|
|
- f'未解决问题应滚动至本周继续跟踪，避免问题"报而不决"形成管理债务。'
|
|
|
- f'建议建立问题看板，按"待处理/处理中/待验证/已关闭"四状态管理。'
|
|
|
- )
|
|
|
- else:
|
|
|
- resolution = (
|
|
|
- f'缺乏上周问题解决率数据。建议建立问题生命周期管理：从上报到关闭全流程记录，'
|
|
|
- f'每周计算"本周关闭率"和"平均关闭时长"。目标：关闭率大于85%、平均时长小于5个工作日。'
|
|
|
- )
|
|
|
- items.append({'title': '✅ 上周问题解决率', 'content': resolution})
|
|
|
-
|
|
|
- # ⑤ 问题预防机制
|
|
|
- prevention = (
|
|
|
- f'建议建立三层预防机制：第一层"系统校验"（如合同超14天自动标红）、'
|
|
|
- f'第二层"流程卡点"（如大额订单必须经法务预审核）、'
|
|
|
- f'第三层"文化驱动"（如每月评选"零问题周"并给予团队奖励）。'
|
|
|
- f'从"救火"转向"防火"，是问题管理的终极目标。'
|
|
|
- )
|
|
|
- items.append({'title': '🛡️ 问题预防机制', 'content': prevention})
|
|
|
-
|
|
|
- return items
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-def _insight_weekly_plan(metrics: dict, context: dict) -> list[dict]:
|
|
|
- """Page 9: 目标拆解逻辑、资源匹配、里程碑、风险对冲"""
|
|
|
- items = []
|
|
|
- next_week_goals = metrics.get('next_week_goals', [])
|
|
|
- monthly_target = context.get('monthly_target', 0)
|
|
|
- total_qty = metrics.get('total_qty', 0)
|
|
|
- tracking_orders = metrics.get('tracking_orders', 0)
|
|
|
-
|
|
|
- # ① 目标拆解逻辑
|
|
|
- if next_week_goals and monthly_target:
|
|
|
- total_goal = sum(g.get('number', 0) for g in next_week_goals)
|
|
|
- breakdown = (
|
|
|
- f'下周目标合计{total_goal}项任务，源自月度目标{monthly_target}台的四分之一拆解（{monthly_target/4:.0f}台/周）。'
|
|
|
- f'本周实际完成{total_qty}台，{"高于" if total_qty > monthly_target/4 else "低于"}周均目标，'
|
|
|
- f'下周需{"保持惯性" if total_qty > monthly_target/4 else "加速追赶"}。'
|
|
|
- f'目标拆解应遵循"存量转化保底+新增拓展增量"双轨逻辑，避免仅靠单一来源支撑。'
|
|
|
- )
|
|
|
- else:
|
|
|
- breakdown = (
|
|
|
- f'下周目标共{len(next_week_goals)}项，因缺乏月度目标或本周台数数据，暂无法做比例拆解。'
|
|
|
- f'建议业务部门输入月度目标后，系统自动按"四周均衡"或"前低后高冲刺"模式生成周目标。'
|
|
|
- f'当前可先基于本周{tracking_orders}单设定下周增长5%-10%的软目标。'
|
|
|
- )
|
|
|
- items.append({'title': '🎯 目标拆解逻辑', 'content': breakdown})
|
|
|
-
|
|
|
- # ② 资源匹配
|
|
|
- n_goals = len(next_week_goals)
|
|
|
- n_owners = len(metrics.get('team', {}).get('owners', {})) if isinstance(metrics.get('team'), dict) and 'owners' in metrics.get('team', {}) else len(metrics.get('team', {}))
|
|
|
- pending_ship = metrics.get('pending_shipment', 0)
|
|
|
- pending_pay = metrics.get('pending_payment', 0)
|
|
|
- resource = (
|
|
|
- f'下周需完成{n_goals}项目标，当前团队{n_owners}人，人均承担{n_goals/n_owners if n_owners else 0:.1f}项目标。'
|
|
|
- f'待发运{pending_ship}单、待收款{pending_pay}单为资源消耗大头，需提前协调物流舱位和财务催收人力。'
|
|
|
- f'若目标增量超过团队当前负载20%，建议申请临时支援或外包非核心环节。'
|
|
|
- )
|
|
|
- items.append({'title': '⚙️ 资源匹配', 'content': resource})
|
|
|
-
|
|
|
- # ③ 里程碑
|
|
|
- milestones = []
|
|
|
- for g in next_week_goals[:2]:
|
|
|
- milestones.append(f'{g.get("id", "")}:{g.get("title", "")}（{g.get("number", 0)}）')
|
|
|
- milestone_text = (
|
|
|
- f'下周必须达成节点：{"；".join(milestones) if milestones else "暂无具体里程碑"}。'
|
|
|
- f'里程碑应满足SMART原则：具体到单数/台数、可量化、可验证。'
|
|
|
- f'建议每周一晨会公开承诺里程碑，周五复盘完成率，未完成项需在当日24:00前提交原因和改进措施。'
|
|
|
- )
|
|
|
- items.append({'title': '🚩 里程碑节点', 'content': milestone_text})
|
|
|
-
|
|
|
- # ④ 风险对冲（Plan B）
|
|
|
- risks = context.get('next_week_risks', [])
|
|
|
- if risks:
|
|
|
- plan_b = (
|
|
|
- f'下周已识别风险：{"、".join([r.get("title", "") for r in risks[:3]])}。'
|
|
|
- f'Plan B原则：若A方案因外部因素受阻，48h内切换至备用方案。'
|
|
|
- f'例如：若主船期延误，提前锁定备用船公司；若大客户延迟下单，启动备选客户清单。'
|
|
|
- f'建议每项高风险目标均配置Plan B，并在周初完成资源预置。'
|
|
|
- )
|
|
|
- else:
|
|
|
- plan_b = (
|
|
|
- f'下周风险清单尚未建立。建议基于本周问题和支持需求，推演下周可能出现的3个最大风险场景：'
|
|
|
- f'（1）物流延误导致E阶段积压、（2）大客户账期延迟导致D→E转化受阻、（3）政策变动导致A阶段合同搁置。'
|
|
|
- f'针对每类场景提前设计应对预案，确保目标韧性。'
|
|
|
- )
|
|
|
- items.append({'title': '🛡️ 风险对冲（Plan B）', 'content': plan_b})
|
|
|
-
|
|
|
- # ⑤ 关键依赖项
|
|
|
- dependencies = context.get('next_week_dependencies', [])
|
|
|
- if dependencies:
|
|
|
- dep_text = (
|
|
|
- f'下周目标达成依赖于：{"、".join(dependencies[:3])}。'
|
|
|
- f'关键依赖应提前一周确认状态，避免临时发现资源不到位导致目标悬空。'
|
|
|
- f'建议建立"依赖项红绿灯"机制：周一全绿确认、周三黄灯预警、周五必须全绿放行。'
|
|
|
- )
|
|
|
- else:
|
|
|
- dep_text = (
|
|
|
- f'下周关键依赖项尚未明确。建议梳理：物流舱位确认书、财务收款截止时间、'
|
|
|
- f'生产排期锁定函等关键外部确认，作为目标达成的先决条件。'
|
|
|
- f'内部依赖（如跨部门协作）也应落实到具体责任人和交付时间。'
|
|
|
- )
|
|
|
- items.append({'title': '🔗 关键依赖项', 'content': dep_text})
|
|
|
-
|
|
|
- return items
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-# =============================================================================
|
|
|
-# MONTHLY INSIGHTS
|
|
|
-# =============================================================================
|
|
|
-
|
|
|
-def monthly_insights(page_type: str, metrics: dict, context: dict) -> list[dict]:
|
|
|
- """Dispatch to specific monthly page insight functions."""
|
|
|
- dispatch = {
|
|
|
- 'monthly_overview': _insight_monthly_overview,
|
|
|
- 'monthly_funnel': _insight_monthly_funnel,
|
|
|
- 'monthly_region': _insight_monthly_region,
|
|
|
- 'monthly_country': _insight_monthly_country,
|
|
|
- 'monthly_trend': _insight_monthly_trend,
|
|
|
- 'monthly_team': _insight_monthly_team,
|
|
|
- 'monthly_support': _insight_monthly_support,
|
|
|
- 'monthly_plan': _insight_monthly_plan,
|
|
|
- }
|
|
|
- fn = dispatch.get(page_type)
|
|
|
- return fn(metrics, context) if fn else []
|
|
|
-
|
|
|
-
|
|
|
-def _insight_monthly_overview(metrics: dict, context: dict) -> list[dict]:
|
|
|
- """Page 3: 月度节奏、目标达成率、季节性、年度进度"""
|
|
|
- items = []
|
|
|
- total_contracts = metrics.get('total_contracts', 0)
|
|
|
- total_qty = metrics.get('total_qty', 0)
|
|
|
- shipped_orders = metrics.get('shipped_orders', 0)
|
|
|
- shipped_qty = metrics.get('shipped_qty', 0)
|
|
|
- trend_by_period = metrics.get('trend_by_period', {})
|
|
|
- monthly_target = context.get('monthly_target', 0)
|
|
|
- yoy_total_qty = metrics.get('yoy_total_qty', 0)
|
|
|
- annual_target = context.get('annual_target', 0)
|
|
|
-
|
|
|
- # ① 月度节奏（上中下旬）
|
|
|
- if trend_by_period:
|
|
|
- early = trend_by_period.get('early', 0)
|
|
|
- mid = trend_by_period.get('mid', 0)
|
|
|
- late = trend_by_period.get('late', 0)
|
|
|
- total_period = early + mid + late
|
|
|
- if total_period:
|
|
|
- early_pct = round(early / total_period * 100, 1)
|
|
|
- mid_pct = round(mid / total_period * 100, 1)
|
|
|
- late_pct = round(late / total_period * 100, 1)
|
|
|
- rhythm = (
|
|
|
- f'本月订单节奏：上旬{early_pct}%（{early:.1f}单/日）、中旬{mid_pct}%（{mid:.1f}单/日）、下旬{late_pct}%（{late:.1f}单/日）。'
|
|
|
- f'{"下旬冲刺特征明显，说明团队具备收官能力但前期储备不足" if late > early and late > mid else ""}'
|
|
|
- f'{"上旬开门红，中下旬平稳，月度节奏健康" if early >= mid and mid >= late * 0.8 else ""}'
|
|
|
- f'{"中旬平台期过长，存在阶段性懈怠，需加强月中跟踪" if mid < early * 0.7 else ""}'
|
|
|
- f'。理想节奏为上旬35%-中旬30%-下旬35%，当前{"接近理想" if 25 <= early_pct <= 45 and 25 <= mid_pct <= 40 else "需调整"}。'
|
|
|
- )
|
|
|
- else:
|
|
|
- rhythm = f'本月分旬数据异常，无法计算节奏占比。建议检查日报数据完整性。'
|
|
|
- else:
|
|
|
- rhythm = (
|
|
|
- f'本月累计{total_contracts}单，缺乏上中下旬分旬数据。'
|
|
|
- f'建议将月度按自然旬拆分，识别"月初开门红/月中平台期/月末冲刺"的典型模式。'
|
|
|
- f'节奏分析是预测下月走势和优化资源投放时点的重要依据。'
|
|
|
- )
|
|
|
- items.append({'title': '📅 月度节奏分析', 'content': rhythm})
|
|
|
-
|
|
|
- # ② 目标达成率
|
|
|
- if monthly_target and total_qty:
|
|
|
- achievement = round(total_qty / monthly_target * 100, 1)
|
|
|
- gap = total_qty - monthly_target
|
|
|
- target_text = (
|
|
|
- f'本月实际完成{total_qty}台，目标{monthly_target}台，达成率{achievement}%，'
|
|
|
- f'{"超额" if gap >= 0 else "缺口"}{abs(gap)}台。'
|
|
|
- f'{"达成率超过100%，建议复盘成功因素并固化为标准打法" if achievement >= 100 else ""}'
|
|
|
- f'{"达成率90%-100%，基本达标但无冗余，需确保最后几天无退单" if 90 <= achievement < 100 else ""}'
|
|
|
- f'{"达成率低于90%，缺口较大，需启动紧急补单或下调下月目标" if achievement < 90 else ""}'
|
|
|
- f'。目标达成率应分解到周，每周偏差大于10%时即触发预警。'
|
|
|
- )
|
|
|
- else:
|
|
|
- target_text = (
|
|
|
- f'本月完成{total_qty}台，因未设定月度目标，无法计算达成率。'
|
|
|
- f'建议业务部门在月初输入目标值，系统将自动按周拆解并生成进度看板。'
|
|
|
- f'当前可先以历史月均值为隐性目标，评估相对表现。'
|
|
|
- )
|
|
|
- items.append({'title': '🎯 目标达成率', 'content': target_text})
|
|
|
-
|
|
|
- # ③ 季节性分析（同比）
|
|
|
- if yoy_total_qty:
|
|
|
- yoy_chg = _pct_change(total_qty, yoy_total_qty)
|
|
|
- seasonal = (
|
|
|
- f'本月{total_qty}台，去年同期{yoy_total_qty}台，同比{_fmt_chg_dir(yoy_chg)}{_fmt_pct(yoy_chg)}。'
|
|
|
- f'{"同比加速增长，市场景气度上行或我司市占率提升" if (yoy_chg or 0) > 10 else ""}'
|
|
|
- f'{"同比小幅增长，与行业大盘同步" if 0 <= (yoy_chg or 0) <= 10 else ""}'
|
|
|
- f'{"同比下滑，需警惕行业下行或竞争加剧" if (yoy_chg or 0) < 0 else ""}'
|
|
|
- f'。建议结合行业报告判断：若我司增速高于行业，说明策略有效；若低于行业，则需重新审视产品/定价/渠道。'
|
|
|
- )
|
|
|
- else:
|
|
|
- seasonal = (
|
|
|
- f'本月完成{total_qty}台，缺乏去年同期数据。'
|
|
|
- f'建议建立年度数据档案，以便进行同比分析。同比是剔除季节性干扰的最佳方式，'
|
|
|
- f'尤其在受春节、海外斋月等影响的月份，环比可能失真，同比更具参考价值。'
|
|
|
- )
|
|
|
- items.append({'title': '📊 季节性分析（同比）', 'content': seasonal})
|
|
|
-
|
|
|
- # ④ 年度进度
|
|
|
- if annual_target and total_qty:
|
|
|
- annual_progress = round(total_qty / annual_target * 100, 1)
|
|
|
- month = context.get('month', 1)
|
|
|
- expected_annual = month / 12 * 100
|
|
|
- annual_gap = annual_progress - expected_annual
|
|
|
- annual_text = (
|
|
|
- f'本月贡献{total_qty}台，占年度目标{annual_target}台的{annual_progress}%，'
|
|
|
- f'按时间进度应达{expected_annual:.1f}%，{"领先" if annual_gap >= 0 else "落后"}{abs(annual_gap):.1f}个百分点。'
|
|
|
- f'{"年度进度超前，建议适当储备Q4项目避免年末透支" if annual_gap > 5 else ""}'
|
|
|
- f'{"年度进度滞后，剩余月份需月均追赶" + str(abs(int(annual_gap/100*annual_target/(12-month)))) + "台" if annual_gap < -5 and month < 12 else ""}'
|
|
|
- f'。年度进度是战略资源配置的北极星指标，每季度应做一次资源再平衡。'
|
|
|
- )
|
|
|
- else:
|
|
|
- annual_text = (
|
|
|
- f'本月完成{total_qty}台，因未设定年度目标，无法计算年度进度。'
|
|
|
- f'建议年初输入年度目标，系统自动按月分解并生成进度条。'
|
|
|
- f'年度目标的合理性直接影响月度策略：过高导致团队透支，过低导致资源闲置。'
|
|
|
- )
|
|
|
- items.append({'title': '🗓️ 年度进度', 'content': annual_text})
|
|
|
-
|
|
|
- # ⑤ 交付闭环效率
|
|
|
- if total_contracts:
|
|
|
- ship_pct = round(shipped_orders / total_contracts * 100, 1)
|
|
|
- delivery = (
|
|
|
- f'本月已发运{shipped_orders}单（{shipped_qty}台），占月内合同{ship_pct}%。'
|
|
|
- f'{"发运占比高，交付闭环效率高" if ship_pct > 60 else "发运占比偏低，大量订单滞留中后期阶段"}。'
|
|
|
- f'需区分"本月新签本月发运"（效率极高）与"历史存量本月发运"（清理旧账），'
|
|
|
- f'前者反映流程效率，后者反映历史包袱。建议分别统计两类占比。'
|
|
|
- )
|
|
|
- else:
|
|
|
- delivery = (
|
|
|
- f'本月暂无合同数据，无法计算交付闭环效率。'
|
|
|
- f'建议建立"当月签约-当月发运"比率作为流程效率核心指标，目标值大于30%。'
|
|
|
- )
|
|
|
- items.append({'title': '🚢 交付闭环效率', 'content': delivery})
|
|
|
-
|
|
|
- return items
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-def _insight_monthly_funnel(metrics: dict, context: dict) -> list[dict]:
|
|
|
- """Page 4: 各环节转化率、瓶颈诊断、历史最优、改进路线图"""
|
|
|
- items = []
|
|
|
- status_funnel = metrics.get('status_funnel', {})
|
|
|
- stage_analysis = metrics.get('stage_analysis', {})
|
|
|
- total_contracts = metrics.get('total_contracts', 0)
|
|
|
-
|
|
|
- if not status_funnel or not total_contracts:
|
|
|
- return [
|
|
|
- {'title': '💡 漏斗结构诊断', 'content': '本月暂无漏斗数据。建议完善订单状态字段，以便计算各阶段转化率并识别瓶颈。'},
|
|
|
- ]
|
|
|
-
|
|
|
- # ① 各环节转化率分析
|
|
|
- names = ['合同拟定中', '已锁定合同待付订金', '已付订金待生产', '已生产待付尾款', '已付尾款待发运', '已发运']
|
|
|
- prev_funnel = context.get('prev_status_funnel', {})
|
|
|
- conv_text = '本月漏斗各阶段占比：'
|
|
|
- parts = []
|
|
|
- for name in names:
|
|
|
- pct = status_funnel.get(name, {}).get('pct', 0)
|
|
|
- prev_pct = prev_funnel.get(name, {}).get('pct', 0) if prev_funnel else 0
|
|
|
- chg = _pct_change(pct, prev_pct)
|
|
|
- parts.append(f'{name}{pct}%({_fmt_chg_dir(chg)}{_fmt_pct(chg)})')
|
|
|
- conv_text += '、'.join(parts[:4]) + '。'
|
|
|
- conv_text += '转化率变化反映流程效率波动，建议锁定两个关键转化节点重点监控：A→B（合同锁定）和D→E（尾款回收）。'
|
|
|
- items.append({'title': '💡 各环节转化率分析', 'content': conv_text})
|
|
|
-
|
|
|
- # ② 瓶颈环节深度诊断
|
|
|
- early = stage_analysis.get('early', {}).get('pct', 0)
|
|
|
- mid = stage_analysis.get('mid', {}).get('pct', 0)
|
|
|
- late = stage_analysis.get('late', {}).get('pct', 0)
|
|
|
- bottleneck = (
|
|
|
- f'阶段结构：前期{early}%（合同+锁定）、中期{mid}%（生产）、后期{late}%（待发运+已发运）。'
|
|
|
- f'{"前期占比过高，新增pipeline充足但转化效率低，需重点突破合同审批和定金催收" if early > 40 else ""}'
|
|
|
- f'{"中期占比过高，生产端积压严重，需排查产能瓶颈或生产计划失衡" if mid > 40 else ""}'
|
|
|
- f'{"后期占比过高，交付压力大，需确保物流和报关资源到位" if late > 40 else ""}'
|
|
|
- f'。与行业benchmark对比：理想状态为前期30%-中期35%-后期35%，当前{"接近理想" if 25 <= early <= 40 and 25 <= mid <= 45 else "偏离理想，需专项改进"}。'
|
|
|
- )
|
|
|
- items.append({'title': '📉 瓶颈环节深度诊断', 'content': bottleneck})
|
|
|
-
|
|
|
- # ③ 历史最优水平对比
|
|
|
- hist_best = context.get('hist_best_conversion', {})
|
|
|
- if hist_best:
|
|
|
- best_text = (
|
|
|
- f'历史最优A→B转化率{hist_best.get("a_to_b", "—")}%、D→E转化率{hist_best.get("d_to_e", "—")}%。'
|
|
|
- f'本月表现与最优水平对比，可量化当前流程效率损失。'
|
|
|
- f'差距大于10个百分点时，说明流程存在显著退化，需启动流程再造；'
|
|
|
- f'差距小于5个百分点时，可通过微调优化逼近最优。建议每季度更新历史最优基准。'
|
|
|
- )
|
|
|
- else:
|
|
|
- best_text = (
|
|
|
- f'本月A→B及D→E转化率数据已记录，但历史最优基准尚未建立。'
|
|
|
- f'建议追溯过去12个月数据，提取各阶段转化率峰值作为"历史最优"标杆。'
|
|
|
- f'历史最优不仅是目标，更是证明"我们曾做到过"的证据，对团队信心建设至关重要。'
|
|
|
- )
|
|
|
- items.append({'title': '🏆 历史最优水平对比', 'content': best_text})
|
|
|
-
|
|
|
- # ④ 改进路线图
|
|
|
- roadmap = (
|
|
|
- f'下月漏斗改进重点：'
|
|
|
- f'（1）A→B转化：优化合同模板，将标准合同审批周期从5天压缩至3天；'
|
|
|
- f'（2）C→D转化：建立生产进度可视化看板，让客户实时追踪车辆生产状态，减少催单焦虑；'
|
|
|
- f'（3）D→E转化：财务前置介入，车辆下线前7天启动尾款提醒，而非传统下线后催收。'
|
|
|
- f'每项改进需设定量化目标、责任人和验收标准，月底复盘执行效果。'
|
|
|
- )
|
|
|
- items.append({'title': '🛣️ 改进路线图', 'content': roadmap})
|
|
|
-
|
|
|
- # ⑤ 资金占用与周转
|
|
|
- pending_pay = metrics.get('pending_payment', {}).get('orders', 0)
|
|
|
- pending_pay_qty = metrics.get('pending_payment', {}).get('qty', 0)
|
|
|
- pending_ship = metrics.get('pending_shipment', {}).get('orders', 0)
|
|
|
- capital = (
|
|
|
- f'本月D阶段（已生产待付尾款）{pending_pay}单（约{pending_pay_qty}台），是资金占用核心环节。'
|
|
|
- f'假设单台均价15万，则D阶段资金占用约¥{pending_pay_qty * 15:,}万。'
|
|
|
- f'若D阶段平均滞留天数超过14天，建议对超期订单启动"财务+销售"联合催收机制，'
|
|
|
- f'目标是将D阶段平均周转天数压降至10天以内。'
|
|
|
- )
|
|
|
- items.append({'title': '💰 资金占用与周转', 'content': capital})
|
|
|
-
|
|
|
- return items
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-def _insight_monthly_region(metrics: dict, context: dict) -> list[dict]:
|
|
|
- """Page 5: 区域战略优先级矩阵、资源投入ROI、订单结构差异"""
|
|
|
- items = []
|
|
|
- region_dist = metrics.get('region_dist', {})
|
|
|
- prev_region_dist = context.get('prev_region_dist', {})
|
|
|
-
|
|
|
- if not region_dist:
|
|
|
- return [
|
|
|
- {'title': '💡 区域战略优先级矩阵', 'content': '本月暂无区域分布数据。建议完善国家-区域映射表，以便按区域维度分析市场规模与增速矩阵。'},
|
|
|
- ]
|
|
|
-
|
|
|
- # ① 区域战略优先级矩阵（市场规模×增速）
|
|
|
- regions = []
|
|
|
- for name, data in region_dist.items():
|
|
|
- qty = data.get('qty', 0)
|
|
|
- pct = data.get('pct', 0)
|
|
|
- prev_qty = prev_region_dist.get(name, {}).get('qty', 0) if prev_region_dist else 0
|
|
|
- growth = _pct_change(qty, prev_qty)
|
|
|
- regions.append((name, qty, pct, growth))
|
|
|
-
|
|
|
- stars = [r for r in regions if r[2] > 20 and (r[3] is None or r[3] > 0)]
|
|
|
- cows = [r for r in regions if r[2] > 20 and (r[3] is not None and r[3] <= 0)]
|
|
|
- questions = [r for r in regions if r[2] <= 20 and (r[3] is None or r[3] > 0)]
|
|
|
- dogs = [r for r in regions if r[2] <= 20 and (r[3] is not None and r[3] <= 0)]
|
|
|
-
|
|
|
- matrix = (
|
|
|
- f'基于"市场规模×增速"四象限分析：'
|
|
|
- f'{"明星（高规模+高增长）：" + "、".join([r[0] for r in stars[:2]]) + "，应追加资源扩大优势；" if stars else ""}'
|
|
|
- f'{"现金牛（高规模+低增长）：" + "、".join([r[0] for r in cows[:2]]) + "，维持投入收割利润；" if cows else ""}'
|
|
|
- f'{"问题（低规模+高增长）：" + "、".join([r[0] for r in questions[:2]]) + "，需判断是真潜力还是虚假繁荣；" if questions else ""}'
|
|
|
- f'{"瘦狗（低规模+低增长）：" + "、".join([r[0] for r in dogs[:2]]) + "，收缩资源或调整模式。" if dogs else ""}'
|
|
|
- )
|
|
|
- items.append({'title': '💡 区域战略优先级矩阵', 'content': matrix})
|
|
|
-
|
|
|
- # ② 资源投入ROI
|
|
|
- roi_text = (
|
|
|
- f'区域ROI评估应超越订单量，引入利润率、售后成本、物流复杂度三维度。'
|
|
|
- f'例如：某区域订单量大但物流成本占比高（偏远国/内陆国），其真实ROI可能低于订单量小的近海区域。'
|
|
|
- f'建议每季度输出区域损益表，按"净利润=订单毛利-物流成本-售后成本-人力分摊"公式计算，'
|
|
|
- f'将资源向高ROI区域倾斜，低ROI区域探索代理商模式降低成本。'
|
|
|
- )
|
|
|
- items.append({'title': '💰 资源投入ROI', 'content': roi_text})
|
|
|
-
|
|
|
- # ③ 区域间订单结构差异
|
|
|
- if len(regions) >= 2:
|
|
|
- r1_name, r1_data = max(region_dist.items(), key=lambda x: x[1].get('qty', 0))
|
|
|
- r1_top = [c['country'] for c in r1_data.get('top_countries', [])[:2]]
|
|
|
- structure = (
|
|
|
- f'{r1_name}为最大区域，其Top国家{r1_top}的车型偏好可能与其他区域存在显著差异。'
|
|
|
- f'例如：亚洲市场偏好紧凑型电动车，非洲市场偏好皮卡/商用车，拉美市场对SUV需求旺盛。'
|
|
|
- f'区域间车型差异反映市场需求本质不同，建议按区域定制产品组合和营销话术，'
|
|
|
- f'避免全球统一策略导致的水土不服。'
|
|
|
- )
|
|
|
- else:
|
|
|
- structure = (
|
|
|
- f'本月区域数据不足，无法分析订单结构差异。'
|
|
|
- f'建议积累多区域数据后，按"区域×车型"交叉分析，识别区域专属需求特征。'
|
|
|
- )
|
|
|
- items.append({'title': '📊 区域间订单结构差异', 'content': structure})
|
|
|
-
|
|
|
- # ④ 区域培育策略
|
|
|
- if questions:
|
|
|
- q_names = [r[0] for r in questions[:2]]
|
|
|
- nurture = (
|
|
|
- f'问题区域（{"、".join(q_names)}）增速快但规模小，处于市场培育期。'
|
|
|
- f'培育策略：前期以"样板工程"为核心，投入1-2个标杆客户确保极致交付体验；'
|
|
|
- f'中期通过客户转介绍和本地车展扩大知名度；后期引入本地代理商实现轻资产扩张。'
|
|
|
- f'培育期通常需要6-12个月，需设定阶段性里程碑避免过早放弃。'
|
|
|
- )
|
|
|
- else:
|
|
|
- nurture = (
|
|
|
- f'本月暂无高增长低基数的问题区域。建议审视现有区域增速：'
|
|
|
- f'若有区域增速超过50%但占比仍低于10%，应立即纳入问题区域清单并给予专项资源。'
|
|
|
- )
|
|
|
- items.append({'title': '🌱 区域培育策略', 'content': nurture})
|
|
|
-
|
|
|
- return items
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-def _insight_monthly_country(metrics: dict, context: dict) -> list[dict]:
|
|
|
- """Page 6: 国家组合健康度、大客户集中度、新国家孵化、竞争态势"""
|
|
|
- items = []
|
|
|
- top_countries = metrics.get('top_countries', {})
|
|
|
- top_countries_change = metrics.get('top_countries_change', {})
|
|
|
- total_qty = metrics.get('total_qty', 0)
|
|
|
-
|
|
|
- if not top_countries:
|
|
|
- return [
|
|
|
- {'title': '💡 国家组合健康度', 'content': '本月暂无国家分布数据。建议完善目的国家字段，以便评估国家组合健康度并制定分散策略。'},
|
|
|
- ]
|
|
|
-
|
|
|
- country_qty = {}
|
|
|
- country_orders = {}
|
|
|
- for c, v in top_countries.items():
|
|
|
- if isinstance(v, int):
|
|
|
- country_qty[c] = v
|
|
|
- country_orders[c] = 0
|
|
|
- else:
|
|
|
- country_qty[c] = v.get('qty', 0)
|
|
|
- country_orders[c] = v.get('orders', 0)
|
|
|
-
|
|
|
- sorted_countries = sorted(country_qty.items(), key=lambda x: x[1], reverse=True)
|
|
|
- top3_qty = sum(v for _, v in sorted_countries[:3])
|
|
|
- top3_pct = round(top3_qty / total_qty * 100, 1) if total_qty else 0
|
|
|
-
|
|
|
- # ① 国家组合健康度评分
|
|
|
- health_score = 100
|
|
|
- if top3_pct > 50:
|
|
|
- health_score -= 30
|
|
|
- elif top3_pct > 40:
|
|
|
- health_score -= 15
|
|
|
- if len(sorted_countries) < 5:
|
|
|
- health_score -= 20
|
|
|
- health = (
|
|
|
- f'国家组合健康度评分{health_score}/100（Top 3集中度{top3_pct}%、覆盖国家数{len(sorted_countries)}）。'
|
|
|
- f'{"评分优秀，国家分布均衡，抗风险能力强" if health_score >= 85 else ""}'
|
|
|
- f'{"评分良好，存在优化空间，建议培育第二梯队" if 70 <= health_score < 85 else ""}'
|
|
|
- f'{"评分偏低，头部依赖严重或覆盖不足，需立即启动分散计划" if health_score < 70 else ""}'
|
|
|
- f'。评分规则：Top 3集中度每超10%扣15分，覆盖国家不足5个扣20分。'
|
|
|
- )
|
|
|
- items.append({'title': '💡 国家组合健康度评分', 'content': health})
|
|
|
-
|
|
|
- # ② 大客户集中度风险
|
|
|
- top1_country, top1_qty = sorted_countries[0]
|
|
|
- top1_orders = country_orders.get(top1_country, 0)
|
|
|
- risk = (
|
|
|
- f'{top1_country}本月{top1_qty}台（{top1_orders}单）领跑。'
|
|
|
- f'若该国最大单一客户占比超过50%，则触发红色预警：该客户流失将导致月度目标缺口{round(top1_qty * 0.5)}台。'
|
|
|
- f'建议对该国客户结构进行"金字塔"分层：底部散户（小于5台）占60%、腰部客户（5-20台）占30%、顶部大客户（大于20台）不超过10%，'
|
|
|
- f'确保任何单一客户流失不会动摇基本盘。'
|
|
|
- )
|
|
|
- items.append({'title': '👤 大客户集中度风险', 'content': risk})
|
|
|
-
|
|
|
- # ③ 新国家孵化进展
|
|
|
- new_countries = context.get('new_countries_this_month', [])
|
|
|
- if new_countries:
|
|
|
- new_text = (
|
|
|
- f'本月新开拓国家：{"、".join(new_countries[:5])}。'
|
|
|
- f'新国家首单是0到1的突破，意义重大但风险较高。建议对新国家实施"护航计划"：'
|
|
|
- f'首单交付全程由资深负责人跟进，确保零差错；交付后30天内回访客户，收集反馈并建立口碑案例；'
|
|
|
- f'若首单反馈良好，第2-3单可逐步放宽管控，交由本地负责人常规跟进。'
|
|
|
- )
|
|
|
- else:
|
|
|
- new_text = (
|
|
|
- f'本月无新国家突破。建议审视"新国家孵化 pipeline"：是否有正在洽谈的潜在市场？'
|
|
|
- f'若连续3个月无新国家，需反思是市场选择过于保守还是孵化流程存在断点。'
|
|
|
- f'建议每季度设定"新国家突破"为团队OKR之一，激励市场拓展。'
|
|
|
- )
|
|
|
- items.append({'title': '🌱 新国家孵化进展', 'content': new_text})
|
|
|
-
|
|
|
- # ④ 竞争态势
|
|
|
- competitive = (
|
|
|
- f'竞争态势监测建议：对Top 3国家建立"竞品对标"机制，跟踪维度包括：'
|
|
|
- f'（1）价格带：竞品是否发起价格战、促销力度如何；'
|
|
|
- f'（2）交付周期：竞品从签约到发运的时长是否短于我方；'
|
|
|
- f'（3）服务网络：竞品是否在当地设立配件仓或服务站。'
|
|
|
- f'本月若某国订单增速放缓但询盘量上升，可能是竞品截流信号，需立即调整策略。'
|
|
|
- )
|
|
|
- items.append({'title': '🏁 竞争态势', 'content': competitive})
|
|
|
-
|
|
|
- return items
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-def _insight_monthly_trend(metrics: dict, context: dict) -> list[dict]:
|
|
|
- """Page 7: 阶段性特征、异常波动归因、外部因素、下月预测"""
|
|
|
- items = []
|
|
|
- trend_by_period = metrics.get('trend_by_period', {})
|
|
|
- daily_trend = metrics.get('daily_trend', {})
|
|
|
- peak_dates = metrics.get('peak_dates', [])
|
|
|
- forecast_next = metrics.get('forecast_next', 0)
|
|
|
- total_qty = metrics.get('total_qty', 0)
|
|
|
- total_contracts = metrics.get('total_contracts', 0)
|
|
|
-
|
|
|
- # ① 阶段性特征
|
|
|
- if trend_by_period:
|
|
|
- early = trend_by_period.get('early', 0)
|
|
|
- mid = trend_by_period.get('mid', 0)
|
|
|
- late = trend_by_period.get('late', 0)
|
|
|
- late_chg = trend_by_period.get('late_change_pct')
|
|
|
- phase = (
|
|
|
- f'本月三旬日均：上旬{early:.1f}单、中旬{mid:.1f}单、下旬{late:.1f}单。'
|
|
|
- f'{"上旬开门红，开局强势" if early > mid and early > late else ""}'
|
|
|
- f'{"中旬平台期，需警惕懈怠" if mid < early * 0.8 and mid < late * 0.8 else ""}'
|
|
|
- f'{"下旬冲刺，收官能力强" if late > early and late > mid else ""}'
|
|
|
- f'下旬较中旬{_fmt_chg_dir(late_chg)}{_fmt_pct(late_chg)}。'
|
|
|
- f'阶段性特征反映了团队的节奏控制能力，理想状态为"高开稳走"，避免过度依赖月末冲刺。'
|
|
|
- )
|
|
|
- else:
|
|
|
- phase = (
|
|
|
- f'本月累计{total_contracts}单，缺乏上中下旬分旬数据。'
|
|
|
- f'建议将月度按自然旬拆分，识别"月初开门红/月中平台期/月末冲刺"的典型模式。'
|
|
|
- f'阶段特征分析有助于优化资源投放时点：上旬重签约、中旬重生产、下旬重交付。'
|
|
|
- )
|
|
|
- items.append({'title': '📅 阶段性特征', 'content': phase})
|
|
|
-
|
|
|
- # ② 异常波动事件归因
|
|
|
- if daily_trend and peak_dates:
|
|
|
- peak_vals = [daily_trend.get(d, 0) for d in peak_dates]
|
|
|
- avg_val = sum(daily_trend.values()) / len(daily_trend)
|
|
|
- anomaly = (
|
|
|
- f'本月峰值日：{"、".join(peak_dates)}（{"、".join([str(v) for v in peak_vals])}单），'
|
|
|
- f'分别为均值的{_safe_div(max(peak_vals), avg_val):.1f}倍。'
|
|
|
- f'峰值日需区分"机会型"（大客户集中下单/展会签约）与"补录型"（历史订单系统批量导入）。'
|
|
|
- f'建议对峰值日逐单标注事件类型，长期积累后可识别真正的业务脉冲与数据噪声。'
|
|
|
- )
|
|
|
- else:
|
|
|
- anomaly = (
|
|
|
- f'本月缺乏分日峰值数据，无法识别异常波动事件。'
|
|
|
- f'建议建立日报机制后，采用"均值+2倍标准差"规则自动标记异常日，并要求负责人提交事件说明。'
|
|
|
- )
|
|
|
- items.append({'title': '⚠️ 异常波动事件归因', 'content': anomaly})
|
|
|
-
|
|
|
- # ③ 外部因素关联
|
|
|
- fx_change = context.get('fx_change_pct')
|
|
|
- policy_events = context.get('policy_events', [])
|
|
|
- shipping_delay = context.get('shipping_delay_days', 0)
|
|
|
- external = (
|
|
|
- f'外部因素评估：'
|
|
|
- f'{"汇率变动" + _fmt_pct(fx_change) + "，对出口报价竞争力产生" + ("正面" if (fx_change or 0) < 0 else "负面") + "影响；" if fx_change is not None else "暂无汇率数据；"}'
|
|
|
- f'{"本月政策事件：" + "、".join(policy_events[:2]) + "；" if policy_events else "暂无重大政策事件；"}'
|
|
|
- f'{"船期平均延误" + str(shipping_delay) + "天，可能滞后影响客户下单信心" if shipping_delay else "船期正常"}。'
|
|
|
- f'外部因素与订单波动存在1-4周滞后，建议建立外部因素登记簿，量化其对后续月份的影响。'
|
|
|
- )
|
|
|
- items.append({'title': '🌐 外部因素关联', 'content': external})
|
|
|
-
|
|
|
- # ④ 下月预测
|
|
|
- mom_pct = _pct_change(total_contracts, metrics.get('prev_total_contracts', 0))
|
|
|
- if forecast_next:
|
|
|
- forecast = (
|
|
|
- f'Pipeline预测下月交付{forecast_next}台，结合本月{total_contracts}单（MoM {_fmt_pct(mom_pct)}），'
|
|
|
- f'下月订单量预计维持在{int(total_contracts * 0.9)}-{int(total_contracts * 1.15)}单区间。'
|
|
|
- f'{"若外部利好持续，有望突破区间上限" if (mom_pct or 0) > 10 else ""}'
|
|
|
- f'{"若趋势承压，可能回落至区间下限，需提前储备补单方案" if (mom_pct or 0) < 0 else ""}'
|
|
|
- f'。预测准确性应每月复盘，偏差超过20%时校准预测模型参数。'
|
|
|
- )
|
|
|
- else:
|
|
|
- forecast = (
|
|
|
- f'本月完成{total_contracts}单，缺乏pipeline预测数据。'
|
|
|
- f'基于历史趋势，下月预计{int(total_contracts * 0.9)}-{int(total_contracts * 1.1)}单。'
|
|
|
- f'建议建立A+B阶段存量订单与下月签约的转化模型，提升预测准确性。'
|
|
|
- )
|
|
|
- items.append({'title': '🔮 下月预测', 'content': forecast})
|
|
|
-
|
|
|
- # ⑤ 趋势持续性信号
|
|
|
- sustain = (
|
|
|
- f'判断下月趋势持续性的三个信号：'
|
|
|
- f'（1）早期pipeline：A+B阶段存量是否充足，能否支撑下月转化；'
|
|
|
- f'（2）客户活跃度：本月更新进度订单数是否维持高位，反映客户 engagement；'
|
|
|
- f'（3）支持需求趋势：若支持需求逐周下降，说明流程顺畅，订单转化阻力减小。'
|
|
|
- f'三个信号中有两个向好，则下月趋势延续概率大于70%。'
|
|
|
- )
|
|
|
- items.append({'title': '📊 趋势持续性信号', 'content': sustain})
|
|
|
-
|
|
|
- return items
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-def _insight_monthly_team(metrics: dict, context: dict) -> list[dict]:
|
|
|
- """Page 8: 人均效能、团队结构优化、激励效果、下月人员配置"""
|
|
|
- items = []
|
|
|
- team = metrics.get('team', {})
|
|
|
- per_capita_orders = metrics.get('per_capita_orders', 0)
|
|
|
- per_capita_qty = metrics.get('per_capita_qty', 0)
|
|
|
- total_contracts = metrics.get('total_contracts', 0)
|
|
|
- total_qty = metrics.get('total_qty', 0)
|
|
|
-
|
|
|
- if isinstance(team, dict) and 'owners' in team:
|
|
|
- owners = team['owners']
|
|
|
- qty_map = team.get('qty', {})
|
|
|
- else:
|
|
|
- owners = {k: v.get('orders', 0) for k, v in team.items()} if team else {}
|
|
|
- qty_map = {k: v.get('qty', 0) for k, v in team.items()} if team else {}
|
|
|
-
|
|
|
- if not owners:
|
|
|
- return [
|
|
|
- {'title': '💡 人均效能趋势', 'content': '本月暂无负责人数据。建议完善订单归属字段，以便评估团队人效并优化配置。'},
|
|
|
- ]
|
|
|
-
|
|
|
- n_members = len(owners)
|
|
|
- sorted_owners = sorted(owners.items(), key=lambda x: x[1], reverse=True)
|
|
|
- top_owner, top_val = sorted_owners[0]
|
|
|
- prev_per_capita = context.get('prev_per_capita_orders', 0)
|
|
|
- yoy_per_capita = context.get('yoy_per_capita_orders', 0)
|
|
|
-
|
|
|
- # ① 人均效能趋势
|
|
|
- mom_chg = _pct_change(per_capita_orders, prev_per_capita)
|
|
|
- yoy_chg = _pct_change(per_capita_orders, yoy_per_capita)
|
|
|
- efficiency = (
|
|
|
- f'本月人均{per_capita_orders:.1f}单（{per_capita_qty:.0f}台），'
|
|
|
- f'环比{_fmt_chg_dir(mom_chg)}{_fmt_pct(mom_chg)}，同比{_fmt_chg_dir(yoy_chg)}{_fmt_pct(yoy_chg)}。'
|
|
|
- f'{"人均效能双升，团队能力在积累" if (mom_chg or 0) > 0 and (yoy_chg or 0) > 0 else ""}'
|
|
|
- f'{"环比升但同比降，说明短期改善但尚未恢复历史水平" if (mom_chg or 0) > 0 and (yoy_chg or 0) < 0 else ""}'
|
|
|
- f'{"人均效能承压，需排查是市场总量下降还是团队效率退化" if (mom_chg or 0) < 0 and (yoy_chg or 0) < 0 else ""}'
|
|
|
- f'。人均效能是团队健康度的核心指标，建议纳入月度考核并与激励挂钩。'
|
|
|
- )
|
|
|
- items.append({'title': '💡 人均效能趋势', 'content': efficiency})
|
|
|
-
|
|
|
- # ② 团队结构优化建议
|
|
|
- tail_count = sum(1 for _, v in sorted_owners if v < per_capita_orders * 0.5)
|
|
|
- structure = (
|
|
|
- f'本月团队{n_members}人，尾部{tail_count}人产出低于人均50%。'
|
|
|
- f'{"尾部占比高，团队呈金字塔结构，需加强腰部建设" if tail_count > n_members // 3 else "尾部占比低，团队呈橄榄型，结构健康"}。'
|
|
|
- f'优化建议：'
|
|
|
- f'（1）低产出人员：分析是能力问题（培训）还是资源问题（客户/国家重新分配）；'
|
|
|
- f'（2）高产出人员：防止过度依赖，建立AB角备份；'
|
|
|
- f'（3）新入职人员：前3个月以保护期为主，第4个月起按正常人均考核。'
|
|
|
- )
|
|
|
- items.append({'title': '⚖️ 团队结构优化建议', 'content': structure})
|
|
|
-
|
|
|
- # ③ 激励效果评估
|
|
|
- incentive_effect = context.get('incentive_effect', '')
|
|
|
- if incentive_effect:
|
|
|
- incentive = (
|
|
|
- f'本月激励政策效果：{incentive_effect}。'
|
|
|
- f'激励效果评估应区分"增量激励"（新签奖励）与"存量激励"（转化奖励），'
|
|
|
- f'避免团队为拿新签奖而忽视存量转化。建议设置激励上限和平衡系数，'
|
|
|
- f'确保短期激励与长期客户价值不冲突。'
|
|
|
- )
|
|
|
- else:
|
|
|
- incentive = (
|
|
|
- f'本月暂无激励效果量化数据。建议下月实施激励政策时，同步记录政策前后的人均产出变化。'
|
|
|
- f'激励效果=（政策后人均产出-政策前人均产出）/激励总成本，ROI大于1.5说明激励有效。'
|
|
|
- )
|
|
|
- items.append({'title': '🏆 激励效果评估', 'content': incentive})
|
|
|
-
|
|
|
- # ④ 标杆对比与提升路径
|
|
|
- top_gap = top_val - per_capita_orders if per_capita_orders else 0
|
|
|
- best_practice = (
|
|
|
- f'本月团队领跑者{top_owner}完成{top_val}单，超人均{top_gap:.1f}单，是均值{_safe_div(top_val, per_capita_orders) if per_capita_orders else 0:.1f}倍。'
|
|
|
- f'建议将其客户跟进SOP、谈判策略、响应时效提炼为标准流程，通过"老带新"机制复制到全团队。'
|
|
|
- f'同时建立月度技能分享会，让头部负责人分享成交案例和失败教训，缩短新人成长周期。'
|
|
|
- f'对于连续两月低于人均50%的成员，启动一对一辅导计划，30天内无明显改善则考虑调岗。'
|
|
|
- )
|
|
|
- items.append({'title': '🎯 标杆对比与提升路径', 'content': best_practice})
|
|
|
-
|
|
|
- # ⑤ 下月人员配置
|
|
|
- next_month_target = context.get('next_month_target', 0)
|
|
|
- if next_month_target and per_capita_orders:
|
|
|
- required = int(next_month_target / per_capita_orders)
|
|
|
- gap = required - n_members
|
|
|
- staffing = (
|
|
|
- f'下月目标{next_month_target}单，按本月人均{per_capita_orders:.1f}单计算，需{required}人，'
|
|
|
- f'当前{n_members}人，{"缺口" + str(gap) + "人，建议启动招聘或内部调配" if gap > 0 else "冗余" + str(abs(gap)) + "人，可优化至其他业务线" if gap < 0 else "刚好匹配"}。'
|
|
|
- f'若市场处于上升期，建议按目标人数的110%配置，预留20%冗余应对突发需求。'
|
|
|
- )
|
|
|
- else:
|
|
|
- staffing = (
|
|
|
- f'下月人员配置建议：维持现有{n_members}人编制，重点优化结构而非单纯扩编。'
|
|
|
- f'若人均产出连续两月下滑，说明市场或团队出现问题，此时扩编只会稀释效率；'
|
|
|
- f'若人均产出持续上升且超负荷，则扩编是必要且紧迫的。'
|
|
|
- )
|
|
|
- items.append({'title': '👥 下月人员配置', 'content': staffing})
|
|
|
-
|
|
|
- return items
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-def _insight_monthly_support(metrics: dict, context: dict) -> list[dict]:
|
|
|
- """Page 9: 需求类型趋势、高频问题根因、流程优化、SLA达成率"""
|
|
|
- items = []
|
|
|
- support_categories = metrics.get('support_categories', {})
|
|
|
- support_count = metrics.get('support_count', 0)
|
|
|
- support_pct = metrics.get('support_pct', 0)
|
|
|
- prev_support_categories = context.get('prev_support_categories', {})
|
|
|
-
|
|
|
- # ① 需求类型趋势
|
|
|
- if support_categories and prev_support_categories:
|
|
|
- trends = []
|
|
|
- for cat, count in support_categories.items():
|
|
|
- prev = prev_support_categories.get(cat, 0)
|
|
|
- chg = _pct_change(count, prev)
|
|
|
- trends.append((cat, chg))
|
|
|
- trends.sort(key=lambda x: (x[1] or 0), reverse=True)
|
|
|
- fastest = trends[0] if trends else (None, None)
|
|
|
- trend_text = (
|
|
|
- f'本月支持需求共{support_count}项（占订单{support_pct}%）。'
|
|
|
- f'增长最快类别为"{fastest[0]}"（{_fmt_pct(fastest[1])}），'
|
|
|
- f'{"说明该领域存在系统性短板，需优先根治" if (fastest[1] or 0) > 50 else ""}'
|
|
|
- f'{"增速可控，按常规节奏优化即可" if (fastest[1] or 0) <= 50 else ""}'
|
|
|
- f'。建议每月输出支持需求趋势图，识别"慢性增长"与"急性爆发"两类问题，分别用流程优化和应急响应处理。'
|
|
|
- )
|
|
|
- else:
|
|
|
- trend_text = (
|
|
|
- f'本月支持需求共{support_count}项（占订单{support_pct}%）。'
|
|
|
- f'缺乏上月分类数据，无法计算各类别增速。建议建立支持需求分类台账，'
|
|
|
- f'按财务/法务/物流/售后/IT五大类归档，每月对比识别增长最快的类别。'
|
|
|
- )
|
|
|
- items.append({'title': '📈 需求类型趋势', 'content': trend_text})
|
|
|
-
|
|
|
- # ② 高频问题根因
|
|
|
- if support_categories:
|
|
|
- top3 = sorted(support_categories.items(), key=lambda x: x[1], reverse=True)[:3]
|
|
|
- root = (
|
|
|
- f'本月Top 3高频问题：{"、".join([f"{c}({v}项)" for c, v in top3])}。'
|
|
|
- f'高频问题的根因通常可归结为三类：'
|
|
|
- f'（1）流程断点：跨部门协作无明确SLA，导致需求在部门间空转；'
|
|
|
- f'（2）信息孤岛：客户/订单/物流数据分散，重复查询浪费人力；'
|
|
|
- f'（3）能力短板：一线人员对产品/政策理解不足，过度依赖支持部门。'
|
|
|
- f'建议对Top 3问题各做一次5Why分析，找到可系统改进的根因。'
|
|
|
- )
|
|
|
- else:
|
|
|
- root = (
|
|
|
- f'本月无支持需求分类数据。建议强制要求提交支持需求时选择类别并简述根因，'
|
|
|
- f'否则不予处理。数据质量是分析的前提，没有分类的数据无法产生洞察。'
|
|
|
- )
|
|
|
- items.append({'title': '🔍 高频问题根因', 'content': root})
|
|
|
-
|
|
|
- # ③ 流程优化建议
|
|
|
- optimization = (
|
|
|
- f'流程优化建议：'
|
|
|
- f'（1）自助化：将Top 20%高频问题转化为FAQ或系统自助查询，减少人工支持需求；'
|
|
|
- f'（2）前置化：在订单关键节点（如合同锁定/生产完成）自动触发检查清单，提前消除潜在问题；'
|
|
|
- f'（3）标准化：对剩余80%中频问题建立SOP和处理模板，将平均处理时长压缩50%。'
|
|
|
- f'优化效果应每月量化：支持需求总量是否下降、重复问题占比是否降低、客户满意度是否提升。'
|
|
|
- )
|
|
|
- items.append({'title': '⚙️ 流程优化建议', 'content': optimization})
|
|
|
-
|
|
|
- # ④ SLA达成率
|
|
|
- sla_data = context.get('sla_achievement', {})
|
|
|
- if sla_data:
|
|
|
- overall = sla_data.get('overall', 0)
|
|
|
- sla_text = (
|
|
|
- f'本月支持需求SLA达成率{overall}%。'
|
|
|
- f'{"达成率大于90%，响应速度优秀" if overall > 90 else "达成率80%-90%，基本达标但存在超期" if overall >= 80 else "达成率低于80%，响应速度严重滞后，需立即增派人手或优化流程"}。'
|
|
|
- f'分类达成率：{"、".join([f"{k}:{v}%" for k, v in sla_data.items() if k != "overall"][:3])}。'
|
|
|
- f'低于目标的类别应作为下月改进重点，分配专项资源提升。'
|
|
|
- )
|
|
|
- else:
|
|
|
- sla_text = (
|
|
|
- f'本月暂无SLA达成率数据。建议为每类支持需求设定处理时限：'
|
|
|
- f'紧急（4h）、高（24h）、中（48h）、低（72h），并记录实际关闭时间。'
|
|
|
- f'SLA是支持部门的服务承诺，也是内部客户体验的核心指标。'
|
|
|
- )
|
|
|
- items.append({'title': '⏱️ SLA达成率', 'content': sla_text})
|
|
|
-
|
|
|
- return items
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-def _insight_monthly_plan(metrics: dict, context: dict) -> list[dict]:
|
|
|
- """Page 10: 目标可行性分析、关键假设、风险场景、Contingency Plan"""
|
|
|
- items = []
|
|
|
- next_month_goals = metrics.get('next_month_goals', [])
|
|
|
- total_qty = metrics.get('total_qty', 0)
|
|
|
- risks = metrics.get('risks', [])
|
|
|
- forecast_next = metrics.get('forecast_next', 0)
|
|
|
-
|
|
|
- # ① 目标可行性分析
|
|
|
- if next_month_goals:
|
|
|
- goal_total = sum(g.get('number', 0) for g in next_month_goals)
|
|
|
- feasibility = (
|
|
|
- f'下月目标共{len(next_month_goals)}项，量化指标合计{goal_total}。'
|
|
|
- f'基于本月{total_qty}台及pipeline预测{forecast_next}台，目标可行性{"高" if forecast_next >= goal_total * 0.8 else "中" if forecast_next >= goal_total * 0.5 else "低"}。'
|
|
|
- f'{"Pipeline充足，目标具备充分支撑" if forecast_next >= goal_total else "Pipeline低于目标，需额外拓展新单填补缺口"}。'
|
|
|
- f'可行性分析应每月更新，若连续两月可行性评级为低，需下调目标或追加资源。'
|
|
|
- )
|
|
|
- else:
|
|
|
- feasibility = (
|
|
|
- f'下月目标尚未设定。建议基于本月{total_qty}台和pipeline{forecast_next}台，'
|
|
|
- f'按"保底（本月×0.9）/基准（本月×1.0）/挑战（本月×1.2）"三档设定目标。'
|
|
|
- f'三档目标分别对应不同的资源投入和激励方案，给团队明确的方向感。'
|
|
|
- )
|
|
|
- items.append({'title': '🎯 目标可行性分析', 'content': feasibility})
|
|
|
-
|
|
|
- # ② 关键假设
|
|
|
- assumptions = context.get('key_assumptions', [])
|
|
|
- if assumptions:
|
|
|
- assump_text = (
|
|
|
- f'下月目标达成的关键假设：{"、".join(assumptions[:3])}。'
|
|
|
- f'关键假设是目标可行性的前提条件，任一假设失效都可能导致目标无法达成。'
|
|
|
- f'建议每周Review假设状态：绿色（稳定）、黄色（波动）、红色（失效），红色假设需在48h内启动应对预案。'
|
|
|
- )
|
|
|
- else:
|
|
|
- assump_text = (
|
|
|
- f'下月目标的关键假设尚未明确。建议至少列出3个最关键假设：'
|
|
|
- f'（1）大客户复购率维持本月水平；'
|
|
|
- f'（2）主要船期无大面积延误；'
|
|
|
- f'（3）汇率波动不超过5%。'
|
|
|
- f'假设清单是风险管理的起点，没有假设的目标只是愿望。'
|
|
|
- )
|
|
|
- items.append({'title': '🔑 关键假设', 'content': assump_text})
|
|
|
-
|
|
|
- # ③ 风险场景（最坏情况）
|
|
|
- if risks:
|
|
|
- worst_gap = context.get('worst_case_gap', 0)
|
|
|
- risk_text = (
|
|
|
- f'已识别风险：{"、".join([r.get("title", "") for r in risks[:3]])}。'
|
|
|
- f'最坏情况下，预计月度缺口约{worst_gap}台。'
|
|
|
- f'缺口弥补方案：（1）加速A→B转化，释放存量pipeline；（2）启动紧急促销，刺激短期下单；'
|
|
|
- f'（3）协调生产加班，压缩交付周期以提升客户信心。'
|
|
|
- f'风险场景应每两周更新一次，确保预案与市场变化同步。'
|
|
|
- )
|
|
|
- else:
|
|
|
- risk_text = (
|
|
|
- f'本月风险清单为空。建议基于历史数据和当前pipeline，推演3个最坏场景：'
|
|
|
- f'（1）Top 1客户延迟下单，缺口30%；（2）船期延误2周，E阶段积压导致客户暂停新签；'
|
|
|
- f'（3）目的国进口政策突变，已锁定合同无法执行。每个场景设定触发条件和应对动作。'
|
|
|
- )
|
|
|
- items.append({'title': '⚠️ 风险场景（最坏情况）', 'content': risk_text})
|
|
|
-
|
|
|
- # ④ Contingency Plan
|
|
|
- contingency = (
|
|
|
- f'Contingency Plan原则：当实际进度低于目标70%或关键假设失效时，自动触发应急预案。'
|
|
|
- f'预案内容包括：'
|
|
|
- f'（1）资源重新配置：将低效区域人力调至高效区域；'
|
|
|
- f'（2）目标动态调整：按"保利润/保现金流/保市场份额"优先级重新排序；'
|
|
|
- f'（3）外部资源调用：启动代理商紧急补单、申请总部促销资源、协调备用物流商。'
|
|
|
- f'预案需在月初制定并获管理层批准，确保危机发生时24h内可执行。'
|
|
|
- )
|
|
|
- items.append({'title': '🛡️ Contingency Plan', 'content': contingency})
|
|
|
-
|
|
|
- # ⑤ 里程碑与复盘机制
|
|
|
- milestones = context.get('monthly_milestones', [])
|
|
|
- if milestones:
|
|
|
- mile_text = (
|
|
|
- f'下月关键里程碑：{"、".join(milestones[:3])}。'
|
|
|
- f'里程碑应满足可验证性：有明确交付物、验收人和截止时间。'
|
|
|
- f'建议建立"双周复盘"机制：每月15日和月底对照里程碑，偏差大于20%时触发专项调整。'
|
|
|
- )
|
|
|
- else:
|
|
|
- mile_text = (
|
|
|
- f'下月里程碑尚未设定。建议按"第一周签约冲刺、第二周生产锁定、第三周尾款回收、第四周交付收官"'
|
|
|
- f'设置4个周里程碑，每个里程碑有量化指标和责任人。没有里程碑的月度计划只是方向，不是计划。'
|
|
|
- )
|
|
|
- items.append({'title': '🚩 里程碑与复盘机制', 'content': mile_text})
|
|
|
-
|
|
|
- return items
|
|
|
-
|
|
|
-
|
|
|
-# =============================================================================
|
|
|
-# HELPER FUNCTIONS
|
|
|
-# =============================================================================
|
|
|
-
|
|
|
-def _pct_change(curr, prev):
|
|
|
- if prev and prev != 0:
|
|
|
- return round((curr - prev) / prev * 100, 1)
|
|
|
- return None
|
|
|
-
|
|
|
-
|
|
|
-def _fmt_pct(val):
|
|
|
- if val is None:
|
|
|
- return '—'
|
|
|
- sign = '+' if val >= 0 else ''
|
|
|
- return f'{sign}{val:.1f}%'
|
|
|
-
|
|
|
-
|
|
|
-def _fmt_chg_dir(val):
|
|
|
- if val is None:
|
|
|
- return ''
|
|
|
- return '增加' if val >= 0 else '减少'
|
|
|
-
|
|
|
-
|
|
|
-def _safe_div(a, b):
|
|
|
- return round(a / b, 1) if b else 0
|
kyle