# -*- coding: utf-8 -*- """17040 原始取数 → daqu_parsed.json（schema 与旧版完全一致）。2026-08-19 起数据全走伽利略 17040。""" import json, shutil, datetime K = "/data/workspace/南部大区城市看板" def load(f): d = json.load(open(f, encoding="utf-8")) return d["data"]["result"] if "data" in d else d["result"] city_m, reg_m, reg_d, city_d = (load(f"{K}/raw_{n}.json") for n in ["city_m", "reg_m", "reg_d", "city_d"]) # 天气增量更新（电费异动辅助因素；失败不阻断主流程） try: import subprocess, sys as _sys subprocess.run([_sys.executable, f"{K}/fetch_weather.py", "12"], timeout=150, check=False) except Exception as _e: print("weather update skipped:", _e) # 仅经营奖面板：7月毛利锁管报口径 2660万；8月毛利管报口径至 8/20（17543771.03），8/21 起取线上数据 JULY_GP_MGMT = 26603012 AUG_GP_MGMT_THRU_0820 = 17543771.0299345 # 城市毛利率目标（月度常量，来自播报表；17040 数据集无目标列） def parse_gpm_targets(): rows = [l.rstrip("\n").split("\t") for l in open(f"{K}/_daqu_raw.tsv", encoding="utf-8")] out = {} for i, r in enumerate(rows): if len(r) > 8 and r[2] == "换电用户" and r[1] and r[1] != "城市": cl, cr = r[1], (r[6] if len(r) > 6 and r[6] and r[6] != "城市" else None) for j in range(i, min(i + 30, len(rows))): rr = rows[j] if len(rr) > 10 and rr[4] == "毛利率目标": try: out[cl] = float(rr[5]) except ValueError: pass if cr: try: out[cr] = float(rr[10]) except ValueError: pass break return out GPM_TGT = parse_gpm_targets() OLD = json.load(open(f"{K}/daqu_parsed.json", encoding="utf-8")) shutil.copy(f"{K}/daqu_parsed.json", f"{K}/daqu_parsed_0817.json") def num(v): try: return float(v) if v not in (None, "", "------") else None except (TypeError, ValueError): return None dates = sorted(set(r["data_date_day"] for r in reg_d)) D1, D2 = dates[-1], dates[-2] as_of = f"{D1[:4]}-{D1[4:6]}-{D1[6:]}" DIM = 31 tp_new = int(D1[6:]) / DIM tp_old = OLD["meta"]["time_progress"] cm = {(r["final_city_name"], str(r["data_date_mounth"])): r for r in city_m} rm = {str(r["data_date_mounth"]): r for r in reg_m} rd = {r["data_date_day"]: r for r in reg_d} cd = {} for r in city_d: cd.setdefault(r["final_city_name"], {})[r["data_date_day"]] = r # ---------- 列名映射 ---------- M_REV, M_GP, M_GPM = "zongshouru_shuihou_buhanshoumai", "huandianyewuyunyingmaoli", "huandianyewuyunyingmaolilv" # 日组件只出成本合计+费率；分项用 收入×费率 反推。租金口径=租金-含手补，用月维 手补/租金 比例补偿 ZUJIN_FACTOR = None def _zujin_factor(): global ZUJIN_FACTOR if ZUJIN_FACTOR is None: r8 = rm.get("202608") ph, fz, rev = mget(r8, "pay_hand"), mget(r8, "feilv_zujin"), mget(r8, M_REV) ZUJIN_FACTOR = ph / (rev * fz) if ph and fz and rev else 1.0 return ZUJIN_FACTOR def _fx(r, k): rev, f = num(r.get("zongshouru_shuihou")), num(r.get(k)) return rev * f if rev is not None and f is not None else None def _abs(r, k): v = r.get(k) if v is None: v = r.get(f"SUM({k})") return num(v) def _pick(a, b): return a if a is not None else b def daily_map(r): zj = _abs(r, "pay_hand") zj_fx = _fx(r, "feilv_zujin") return { "date": r["data_date_day"], "users": num(r.get("SUM(huandian_users_ym)") or r.get("huandian_users_ym")), "users_hd": num(r.get("SUM(huandian_hdc_users_ym)") or r.get("huandian_hdc_users_ym")), "kedanjia": num(r.get("kedanjia")), "kedanjia_noka": num(r.get("kedanjia_buhanxianxiaka")), "kwh_user": num(r.get("danyonghurijunyongdianliang_") or r.get("danyonghurijunyongdianliang")), "dudian": num(r.get("dudiandanjia")), "rev": num(r.get("zongshouru_shuihou") or r.get("zongshouru_shuihou_buhanshoumai")), "cost": num(r.get("chengben") or r.get("chengben_buhanshoumai")), "dianfei": _pick(_abs(r, "pay_ele_fee_merchant"), _fx(r, "feilv_dianfei")), "btrydep": _pick(_abs(r, "btry_depre_cost"), _fx(r, "feilv_dianchizhejiu")), "cabtdep": _pick(_abs(r, "cabt_depre_cost"), _fx(r, "feilv_dianguizhejiu")), "zujin": (((zj if zj is not None else (zj_fx or 0)) * _zujin_factor()) if (zj is not None or zj_fx is not None) else None), "renli": _pick(_abs(r, "quanzhirenli"), _fx(r, "feilv_quanzhirenli")), "jianzhi": _fx(r, "feilv_jianzhilaowu"), "daiyunying": _fx(r, "feilv_daiyunyingchengben"), "anzhuang": _pick(_abs(r, "anzhuangchengben"), _fx(r, "feilv_anzhuangchengben")), "qita_cost": _pick(_abs(r, "qitachengben_qita"), _fx(r, "feilv_qitachengben")), "yingxiao": _pick(_abs(r, "yxf_total"), _fx(r, "feilv_zongyingxiaofei")), "gp": num(r.get("huandianyewuyunyingmaoli")), "gpm": num(r.get("huandianyewuyunyingmaolilv")), "gp_user": num(r.get("danyonghuyunyingmaoli")), } # 月维度的成本列名带 SUM() 前缀的兼容 def mget(r, k): if r is None: return None for key in (k, f"SUM({k})"): if key in r: return num(r[key]) return None region_daily = [daily_map(rd[d]) for d in dates] # ---------- region ---------- def derive_target(old_mtd, old_gap, old_tp, ratio=False): return (old_mtd - old_gap) if ratio else old_mtd / (old_gap + old_tp) T_REV = derive_target(OLD["region"]["收入"]["mtd"], OLD["region"]["收入"]["gap"], tp_old) T_GP = derive_target(OLD["region"]["毛利额"]["mtd"], OLD["region"]["毛利额"]["gap"], tp_old) T_GMR = OLD["region"]["毛利率"]["mtd"] - OLD["region"]["毛利率"]["gap"] r8, r1d, r2d = rm.get("202608"), rd[D1], rd[D2] def trio(mtd, t1, t2): return {"mtd": mtd, "t1": t1, "dod": (t1 - t2) if t1 is not None and t2 is not None else None} o = OLD["region"] region = { "收入": {**trio(mget(r8, M_REV), mget(r1d, "zongshouru_shuihou"), mget(r2d, "zongshouru_shuihou")), "name": "收入", "gap": mget(r8, M_REV) / T_REV - tp_new}, "毛利额": {**trio(mget(r8, M_GP), mget(r1d, M_GP), mget(r2d, M_GP)), "name": "毛利额", "gap": mget(r8, M_GP) / T_GP - tp_new}, "毛利率": {**trio(mget(r8, M_GPM), mget(r1d, M_GPM), mget(r2d, M_GPM)), "name": "毛利率", "gap": mget(r8, M_GPM) - T_GMR}, "单用户毛利额_本月水位": mget(r8, "danyonghuyunyingmaoli"), } # 联营 = city_type 联营 城市加总 ly_cities = sorted(set(r["final_city_name"] for r in city_m if r.get("city_type") == "联营")) def sum_cities(rows, k): return sum((mget(x, k) or 0) for x in rows) ly8 = [x for x in city_m if x.get("city_type") == "联营" and str(x["data_date_mounth"]) == "202608"] ly1 = [cd[c][D1] for c in ly_cities if c in cd and D1 in cd[c]] ly2 = [cd[c][D2] for c in ly_cities if c in cd and D2 in cd[c]] ly = OLD["region"]["联营毛利额"] region["联营毛利额"] = { "mtd": sum_cities(ly8, M_GP), "t1": sum_cities(ly1, M_GP), "dod": (sum_cities(ly1, M_GP) - sum_cities(ly2, M_GP)) if ly1 and ly2 else None, "target": ly["target"], "gap": sum_cities(ly8, M_GP) / ly["target"] - tp_new, } # ---------- cities（12直营，沿用旧城市清单与静态目标） ---------- cities = {} for c, oldc in OLD["cities"].items(): m8 = cm.get((c, "202608")); d1 = cd.get(c, {}).get(D1); d2 = cd.get(c, {}).get(D2) def cell(km, kd=None, src_m=m8, src1=d1, src2=d2): kd = kd or km t1, t2 = mget(src1, kd), mget(src2, kd) return {"mtd": mget(src_m, km), "t1": t1, "dod": (t1 - t2) if t1 is not None and t2 is not None else None} # 成本结构口径（2026-08-19 用户确认）：租金=租金-含手补；电费=电费成本； # 全职人力&资产折旧=人力成本+电池折旧+电柜折旧；安装成本=安装成本； # 其他成本=成本合计-租金-电费-全职人力&资产折旧-安装成本 HR_K = ["quanzhirenli", "btry_depre_cost", "cabt_depre_cost"] def _s(r, ks=HR_K): if r is None: return None vs = [mget(r, k) for k in ks] return sum(v or 0 for v in vs) if any(v is not None for v in vs) else None def _qita(t, z, f, h, a): if any(v is None for v in (t, z, f, h, a)): return None return t - z - f - h - a hr_m, hr_1, hr_2 = _s(m8), _s(d1), _s(d2) tot_m, tot_1, tot_2 = mget(m8, "chengben_buhanshoumai"), mget(d1, "chengben"), mget(d2, "chengben") zj_m, zj_1, zj_2 = mget(m8, "pay_hand"), mget(d1, "pay_hand"), mget(d2, "pay_hand") df_m, df_1, df_2 = mget(m8, "pay_ele_fee_merchant"), mget(d1, "pay_ele_fee_merchant"), mget(d2, "pay_ele_fee_merchant") az_m, az_1, az_2 = mget(m8, "anzhuangchengben"), mget(d1, "anzhuangchengben"), mget(d2, "anzhuangchengben") qita_m, qita_1, qita_2 = (_qita(tot_m, zj_m, df_m, hr_m, az_m), _qita(tot_1, zj_1, df_1, hr_1, az_1), _qita(tot_2, zj_2, df_2, hr_2, az_2)) old_tu = oldc.get("换电用户目标") or {"mtd": None, "t1": None, "dod": None} old_tg = oldc.get("毛利额目标") or {"mtd": None, "t1": None, "dod": None} tgt_gp = old_tg["mtd"] if isinstance(old_tg, dict) else old_tg gp_mtd = mget(m8, M_GP) vs = (gp_mtd / tgt_gp - tp_new) if gp_mtd is not None and tgt_gp else None rev_mtd = mget(m8, M_REV); hd_rev = mget(m8, "huandianshouru") cities[c] = { "换电用户": {"mtd": mget(m8, "huandian_users_ym"), "t1": mget(d1, "huandian_users_ym"), "dod": (mget(d1, "huandian_users_ym") - mget(d2, "huandian_users_ym")) if d1 and d2 else None}, "换电用户目标": old_tu, "换电用户含单次": {"mtd": mget(m8, "huandian_hdc_users_ym"), "t1": None, "dod": None}, "收入": cell(M_REV, "zongshouru_shuihou"), "换电收入": cell("huandianshouru"), "其他收入": {"mtd": (rev_mtd - hd_rev) if rev_mtd is not None and hd_rev is not None else None, "t1": (mget(d1, "zongshouru_shuihou") - mget(d1, "huandianshouru")) if d1 else None, "dod": None}, "客单价（去KA）": {"mtd": mget(m8, "kedanjia_buhanxianxiaka"), "t1": mget(d1, "kedanjia_buhanxianxiaka"), "dod": (mget(d1, "kedanjia_buhanxianxiaka") - mget(d2, "kedanjia_buhanxianxiaka")) if d1 and d2 else None}, "成本": cell("chengben_buhanshoumai", "chengben"), "租金": cell("pay_hand"), "电费": cell("pay_ele_fee_merchant"), "-电费费率": cell("feilv_dianfei"), "-单用户用电量": cell("danyonghurijunyongdianliang_", "danyonghurijunyongdianliang"), "全职人力&资产折旧": {"mtd": hr_m, "t1": hr_1, "dod": (hr_1 - hr_2) if hr_1 is not None and hr_2 is not None else None}, "安装成本": cell("anzhuangchengben"), "其他成本": {"mtd": qita_m, "t1": qita_1, "dod": (qita_1 - qita_2) if qita_1 is not None and qita_2 is not None else None}, "营销费": cell("yxf_total"), "-营销费率": cell("feilv_zongyingxiaofei"), "本月单用户毛利额": cell("danyonghuyunyingmaoli"), "毛利额": cell(M_GP), "毛利率": cell(M_GPM), "毛利率目标": GPM_TGT.get(c), "毛利额目标": old_tg, "毛利额GAP": (gp_mtd - tgt_gp * tp_new) if gp_mtd is not None and tgt_gp else None, # 按时间进度折算：MTD − 目标×(日/31) "月度目标达成进度VS时间进度": vs, } # ---------- award ---------- old_aw = OLD["award"] gp7 = JULY_GP_MGMT if D1 >= "20260820": aug_after_0820 = sum(daily_map(rd[d]).get("gp") or 0 for d in dates if "20260821" <= d <= D1) gp8 = AUG_GP_MGMT_THRU_0820 + aug_after_0820 # 8月=管报(≤8/20)+线上(8/21起) else: gp8 = mget(rm.get("202608"), M_GP) award = { "q_targets": old_aw["q_targets"], "users_target_avg": old_aw["users_target_avg"], "achieve_gp": {"202607": gp7, "202608": gp8, "合计": (gp7 or 0) + (gp8 or 0)}, "gap_gp": {"202607": (gp7 or 0) - old_aw["q_targets"]["202607"], "202608": (gp8 or 0) - old_aw["q_targets"]["202608"]}, "target_gap": (gp7 or 0) + (gp8 or 0) - old_aw["q_targets"]["合计"], "start_line": old_aw["start_line"], "achieved": old_aw.get("achieved"), "time_progress": old_aw["time_progress"], # gen_html2 会按日历重算 "rest_daily": old_aw.get("rest_daily"), "excess_gap": old_aw.get("excess_gap"), "est_award": old_aw.get("est_award"), } # ---------- city_daily ---------- city_daily = {} for c in OLD["city_daily"]: rows = cd.get(c, {}) city_daily[c] = [daily_map(rows[d]) for d in sorted(rows)] # ---------- city_ext：趋势对比 / 月底预估 / 异常标签 / 电费异动（全城市） ---------- try: WEATHER = json.load(open(f"{K}/weather_city.json", encoding="utf-8")) except Exception: WEATHER = {} def _avg(vs): vs = [v for v in vs if v is not None] return sum(vs) / len(vs) if vs else None CMP_KEYS = ["收入", "毛利额", "毛利率", "换电用户", "单用户收入", "电费率"] RATE_KEYS = {"毛利率", "电费率"} def _metrics_of(r): rev, users, df = r.get("rev"), r.get("users"), r.get("dianfei") return { "收入": rev, "毛利额": r.get("gp"), "毛利率": r.get("gpm"), "换电用户": users, "单用户收入": (rev / users) if rev and users else None, "电费率": (df / rev) if rev and df is not None else None, } def _cmp(series): """6 指标 × 3 对比：日环比(T-1 vs T-2) / 较上周同日(T-1 vs T-8) / 近7日周环比(7日均 vs 前7日均)。率类为 pp 差。""" if len(series) < 14: return None t1, t2, t8 = series[-1], series[-2], series[-8] l7, p7 = series[-7:], series[-14:-7] l7a = {k: _avg([_metrics_of(r)[k] for r in l7]) for k in CMP_KEYS} p7a = {k: _avg([_metrics_of(r)[k] for r in p7]) for k in CMP_KEYS} out = {} for k in CMP_KEYS: a1, a2, a8 = _metrics_of(t1)[k], _metrics_of(t2)[k], _metrics_of(t8)[k] def _d(a, b): if a is None or b is None: return None return (a - b) if k in RATE_KEYS else ((a / b - 1) if b else None) out[k] = {"t1": a1, "dod": _d(a1, a2), "wowsame": _d(a1, a8), "w7": _d(l7a[k], p7a[k])} return out def _fc(c, s): """月底预估：收入=近7日run-rate；变动成本(电费)=预估收入×近7日电费费率； 固定成本(租金+人力折旧+其他)=按天线性计提；一次性成本(安装)=MTD实际不计提；营销费=run-rate。""" v = cities.get(c) if not v or len(s) < 7: return None elapsed, remain = int(D1[6:8]), DIM - int(D1[6:8]) if remain <= 0 or elapsed <= 0: return None l7 = s[-7:] rev_mtd = (v.get("收入") or {}).get("mtd") rev_l7, mkt_l7 = _avg([r["rev"] for r in l7]), _avg([r["yingxiao"] or 0 for r in l7]) if rev_mtd is None or not rev_l7: return None rev_proj = rev_mtd + rev_l7 * remain df_sum, rev_sum = sum(r["dianfei"] or 0 for r in l7), sum(r["rev"] or 0 for r in l7) ele_rate = df_sum / rev_sum if rev_sum else None var_proj = rev_proj * ele_rate if ele_rate is not None else None g = lambda k: (v.get(k) or {}).get("mtd") fixed_mtd = sum(x for x in [g("租金"), g("全职人力&资产折旧"), g("其他成本")] if x is not None) fixed_proj = fixed_mtd / elapsed * DIM onetime = g("安装成本") or 0 mkt_proj = (g("营销费") or 0) + mkt_l7 * remain if var_proj is None: return None cost_proj = var_proj + fixed_proj + onetime + mkt_proj gp_proj = rev_proj - cost_proj gpm_proj = gp_proj / rev_proj if rev_proj else None gpm_tgt = v.get("毛利率目标") return { "rev_proj": rev_proj, "gp_proj": gp_proj, "gpm_proj": gpm_proj, "gpm_tgt": gpm_tgt, "gpm_ach": (gpm_proj / gpm_tgt) if gpm_proj is not None and gpm_tgt else None, "gp_gap_proj": (gp_proj - rev_proj * gpm_tgt) if gpm_tgt is not None else None, "ele_rate_l7": ele_rate, "parts": {"变动成本(电费)预估": var_proj, "固定成本计提": fixed_proj, "一次性成本(安装)按实际": onetime, "营销费预估": mkt_proj}, "methods": {"变动": "预估收入×近7日电费费率", "固定": "租金+人力折旧+其他 按天线性计提", "一次性": "安装成本按MTD实际、不外推", "营销": "近7日run-rate"}, } def _ele_decomp(s, c): """电费率变化三因子分解（近7日 vs 前7日）：电费率 = 单用户用电量×度电单价/单用户收入。 顺序替代法，三项之和 = Δrate(重构)。天气仅作辅助相关因素展示。""" if len(s) < 14: return None def agg(seg): rev = sum(r["rev"] or 0 for r in seg) users = _avg([r["users"] for r in seg]) k, p = _avg([r["kwh_user"] for r in seg]), _avg([r["dudian"] for r in seg]) df = sum(r["dianfei"] or 0 for r in seg) rpu = rev / 7 / users if users else None return {"rev": rev, "users": users, "k": k, "p": p, "rpu": rpu, "rate": df / rev if rev else None, "df": df} a0, a1 = agg(s[-14:-7]), agg(s[-7:]) out = {"a0": a0, "a1": a1, "d_usage": None, "d_price": None, "d_denom": None, "rate_star0": None, "rate_star1": None} if all(x is not None for x in (a0["k"], a0["p"], a0["rpu"], a1["k"], a1["p"], a1["rpu"])): k0, p0, r0, k1, p1, r1 = a0["k"], a0["p"], a0["rpu"], a1["k"], a1["p"], a1["rpu"] out["rate_star0"], out["rate_star1"] = k0 * p0 / r0, k1 * p1 / r1 out["d_usage"] = (k1 - k0) * p0 / r0 out["d_price"] = k1 * (p1 - p0) / r0 out["d_denom"] = k1 * p1 * (1 / r1 - 1 / r0) # 天气（辅助）：近7日 vs 前7日 dates_l7 = [r["date"] for r in s[-7:]] dates_p7 = [r["date"] for r in s[-14:-7]] def wagg(ds): t = [WEATHER.get(c, {}).get(d, {}).get("t") for d in ds] ra = [WEATHER.get(c, {}).get(d, {}).get("r") for d in ds] return {"temp": _avg(t), "rain": sum(x or 0 for x in ra) if any(x is not None for x in ra) else None} out["weather"] = {"l7": wagg(dates_l7), "p7": wagg(dates_p7)} # 异常日：近7日中电费率高于7日均值1pp 或 日环比+1pp l7r = [(r["date"], (r["dianfei"] / r["rev"]) if r["rev"] and r["dianfei"] is not None else None) for r in s[-7:]] base = _avg([x for _, x in l7r if x is not None]) prev_rate = None if len(s) >= 8 and s[-8]["rev"] and s[-8]["dianfei"] is not None: prev_rate = s[-8]["dianfei"] / s[-8]["rev"] anom = [] for d, rate in l7r: if rate is None or base is None: continue flag = (rate - base > 0.01) or (prev_rate is not None and rate - prev_rate > 0.01) if flag: w = WEATHER.get(c, {}).get(d, {}) anom.append({"date": d, "rate": rate, "temp": w.get("t"), "rain": w.get("r")}) prev_rate = rate out["anom_days"] = anom # 14 天序列（图用） out["series"] = [] for r in s[-14:]: w = WEATHER.get(c, {}).get(r["date"], {}) out["series"].append({"date": r["date"], "rate": (r["dianfei"] / r["rev"]) if r["rev"] and r["dianfei"] is not None else None, "kwh": r["kwh_user"], "p": r["dudian"], "temp": w.get("t"), "rain": w.get("r")}) return out def _tags(c, cmp_, v, fc, ele): t = [] if cmp_: if cmp_["收入"]["dod"] is not None and cmp_["收入"]["dod"] < -0.08: t.append(("收入骤降", "r")) if cmp_["换电用户"]["w7"] is not None and cmp_["换电用户"]["w7"] < -0.03: t.append(("用户流失", "r")) if cmp_["单用户收入"]["w7"] is not None and cmp_["单用户收入"]["w7"] < -0.03: t.append(("单用户收入下滑", "o")) if ele and ele["a1"]["rate"] is not None and ele["a1"]["rate"] > 0.30: t.append(("电费费率超30%", "r")) if v: gpm, tgt = (v.get("毛利率") or {}).get("mtd"), v.get("毛利率目标") if gpm is not None and tgt is not None and gpm < tgt: t.append(("毛利率不达标", "r" if tgt - gpm > 0.02 else "o")) vs = v.get("月度目标达成进度VS时间进度") if vs is not None and vs < -0.02: t.append(("进度落后", "o")) return t[:4] city_ext = {} # 月维 MTD 收入/毛利率（所有城市，供风险矩阵用） _mtd_rev, _mtd_gpm = {}, {} for r in city_m: if str(r.get("data_date_mounth")) == D1[:6]: c = r["final_city_name"] _mtd_rev[c] = mget(r, M_REV) _mtd_gpm[c] = mget(r, M_GPM) for c in sorted({r["final_city_name"] for r in city_m}): rows = cd.get(c, {}) if not rows: continue s = [daily_map(rows[d]) for d in sorted(rows)] cmp_ = _cmp(s) fc = _fc(c, s) if c in cities else None ele = _ele_decomp(s, c) city_ext[c] = { "cmp": cmp_, "fc": fc, "ele": ele, "tags": _tags(c, cmp_, cities.get(c), fc, ele), "spark7": [round((r["gp"] or 0) / 1e4, 2) for r in s[-7:]], "type": next((r.get("city_type") for r in city_m if r["final_city_name"] == c), None), "rev_mtd": _mtd_rev.get(c) or 0, "gpm_mtd": _mtd_gpm.get(c), "gpm_tgt": cities.get(c, {}).get("毛利率目标") if c in cities else None, } # ---------- 大区城市变化贡献排名（T-1 vs T-2） ---------- contrib = {} for mk, dk in [("收入", "rev"), ("毛利额", "gp")]: deltas = [] for c, rows in cd.items(): if D1 not in rows or D2 not in rows: continue a, b = daily_map(rows[D1]).get(dk), daily_map(rows[D2]).get(dk) if a is not None and b is not None: deltas.append({"city": c, "delta": a - b}) tot = sum(x["delta"] for x in deltas) for x in deltas: x["share"] = x["delta"] / tot if tot else None contrib[mk] = sorted(deltas, key=lambda x: x["delta"]) # ---------- 经营风险矩阵（毛利率缺口 × 收入周环比 × 规模） ---------- risk = [] for c, e in city_ext.items(): if not (e["rev_mtd"] or 0) > 0: continue gpm = e["gpm_mtd"] tgt = e["gpm_tgt"] if e["gpm_tgt"] else 0.25 # 无目标城市按 25% 参考线 w7 = (e["cmp"] or {}).get("收入", {}).get("w7") if e["cmp"] else None risk.append({"city": c, "type": e["type"], "rev": e["rev_mtd"], "x": (gpm - tgt) if gpm is not None else None, "y": w7, "has_tgt": bool(e["gpm_tgt"])}) meta = {"title": OLD["meta"]["title"], "as_of": as_of, "mtd_end": D1, "days_in_month": DIM, "time_progress": tp_new} out = {"meta": meta, "region": region, "award": award, "cities": cities, "region_daily": region_daily, "city_daily": city_daily, "city_ext": city_ext, "contrib": contrib, "risk": risk} json.dump(out, open(f"{K}/daqu_parsed.json", "w", encoding="utf-8"), ensure_ascii=False) # ---------- prep_data 输入 & extra_data ---------- shutil.copy(f"{K}/../.kiro/cache/read-polaris-skill/tmp/south_city_monthly_v2.json", f"{K}/../.kiro/cache/read-polaris-skill/tmp/south_city_monthly_v2.json.bak_0817") json.dump(city_m, open("/data/workspace/.kiro/cache/read-polaris-skill/tmp/south_city_monthly_v2.json", "w", encoding="utf-8"), ensure_ascii=False) json.dump(reg_m, open("/data/workspace/.kiro/cache/read-polaris-skill/tmp/south_monthly_trend.json", "w", encoding="utf-8"), ensure_ascii=False) E = json.load(open(f"{K}/extra_data.json", encoding="utf-8")) E["gp_user_core"] = {"mtd": mget(r8, "danyonghuyunyingmaoli"), "t1": mget(r1d, "danyonghuyunyingmaoli"), "t2": mget(r2d, "danyonghuyunyingmaoli"), "asof": as_of, "source": "经营管理报表_核心指标"} json.dump(E, open(f"{K}/extra_data.json", "w", encoding="utf-8"), ensure_ascii=False) print("as_of:", as_of, "| tp_monthly %.4f" % tp_new, "| region rev mtd %.0f t1 %.0f" % (region["收入"]["mtd"], region["收入"]["t1"])) print("award achieve_gp:", {k: (round(v/1e4) if v else v) for k, v in award["achieve_gp"].items()}) print("联营 mtd %.0f t1 %.0f" % (region["联营毛利额"]["mtd"], region["联营毛利额"]["t1"])) print("region_daily:", len(region_daily), dates[0], "->", D1)