""" 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