classify.py

07-Automation/partner_pipeline/classify.py

1. """classify and verify: tiers from lifetime metrics, and the no-shrink check."""
3. from __future__ import annotations
5. import csv
6. import json
7. import sys
8. from pathlib import Path
10. from .config import is_armed, load_config
11. from .history import (apply_merges, compute_metrics, load_history, load_merge_registry,
12. save_history, to_float)
13. from .paths import OUT_DIR, TODAY
14. from .stubs import index_existing_stubs, stub_status
17. def tier_override_for(c: dict, cfg: dict) -> str | None:
18. """Relationship-set tier, if this mailbox has one (partner-thresholds.md
19. `tier_overrides.emails`). Wins over the computed band and over
20. min_orders_to_qualify: an overridden account holds its tier at $0 spend.
21. Scope is per-email on purpose; see the config comment before widening it."""
22. ov = (cfg.get("tier_overrides") or {}).get("emails") or {}
23. if not ov:
24. return None
25. email = (c.get("email") or "").strip().lower()
26. return ov.get(email) if email else None
29. def classify_one(c: dict, cfg: dict, armed: bool) -> str:
30. if not armed:
31. return "UNCLASSIFIED: thresholds not set"
32. forced = tier_override_for(c, cfg)
33. if forced:
34. return forced
35. min_orders = cfg.get("min_orders_to_qualify") or 0
36. if c["total_orders_count"] < min_orders:
37. return "Not a partner client"
38. mode = cfg.get("mode", "ltv_bands")
40. if mode == "ltv_bands":
41. ltv = c["ltv"]
42. for band in cfg.get("ltv_bands", []):
43. lo, hi = band.get("min_ltv"), band.get("max_ltv")
44. if lo is None:
45. continue
46. if ltv >= lo and (hi is None or ltv < hi):
47. return band["tier"]
48. return "Not a partner client"
50. if mode == "simple_gate":
51. g = cfg.get("simple_gate", {})
52. val = c.get(g.get("signal", "ltv"), 0)
53. return "Partner Client" if val >= (g.get("min_value") or 0) else "Not a partner client"
55. if mode == "multi_factor":
56. mf = cfg.get("multi_factor", {})
57. w = mf.get("weights", {})
58. score = (
59. to_float(w.get("ltv")) * c["ltv"]
60. + to_float(w.get("repeat_revenue")) * c["repeat_revenue"]
61. + to_float(w.get("order_count")) * c["total_orders_count"]
62. )
63. if score < to_float(mf.get("partner_cutoff")):
64. return "Not a partner client"
65. for band in mf.get("score_bands", []) or []:
66. lo, hi = band.get("min_score"), band.get("max_score")
67. if lo is not None and score >= lo and (hi is None or score < hi):
68. return band["tier"]
69. return "Partner Client"
71. return "Not a partner client"
74. COLUMNS = [
75. "user_id", "email", "first_name", "last_name", "company", "phone",
76. "total_orders_count", "ltv", "repeat_revenue", "first_order_date",
77. "last_order_date", "assigned_tier", "store_user_group", "has_baseline",
78. "post_baseline_orders", "stub_status",
79. ]
82. def export_spreadsheet(rows: list[dict], summary: dict) -> Path:
83. OUT_DIR.mkdir(parents=True, exist_ok=True)
84. stem = OUT_DIR / f"partner-classification_{TODAY}"
85. (stem.with_suffix(".plan.json")).write_text(
86. json.dumps({"rows": rows, "summary": summary}, indent=2), encoding="utf-8"
87. )
88. try:
89. from openpyxl import Workbook
90. wb = Workbook()
91. ws = wb.active
92. ws.title = "Classification"
93. ws.append(COLUMNS)
94. for r in rows:
95. ws.append([r.get(k, "") for k in COLUMNS])
96. s = wb.create_sheet("Summary")
97. for k, v in summary.items():
98. s.append([k, json.dumps(v) if isinstance(v, (dict, list)) else v])
99. out = stem.with_suffix(".xlsx")
100. wb.save(out)
101. return out
102. except ImportError:
103. out = stem.with_suffix(".csv")
104. with out.open("w", newline="", encoding="utf-8") as f:
105. w = csv.DictWriter(f, fieldnames=COLUMNS, extrasaction="ignore")
106. w.writeheader()
107. w.writerows(rows)
108. print("NOTE: openpyxl not installed: wrote CSV instead of xlsx.")
109. return out
112. def cmd_classify(args):
113. cfg = load_config()
114. armed, reasons = is_armed(cfg)
115. h = load_history()
116. metrics = compute_metrics(h, cfg)
118. aliases, _ = load_merge_registry()
119. metrics, folded = apply_merges(metrics, aliases)
120. if folded:
121. print(f"MERGE REGISTRY: folded {len(folded)} absorbed account(s): "
122. + ", ".join(folded))
124. idx = index_existing_stubs()
125. rows, tier_counts, status_counts = [], {}, {}
126. for c in metrics.values():
127. tier = classify_one(c, cfg, armed)
128. status, _ = stub_status(c, idx)
129. tier_counts[tier] = tier_counts.get(tier, 0) + 1
130. status_counts[status] = status_counts.get(status, 0) + 1
131. rows.append({**c, "assigned_tier": tier, "stub_status": status})
133. rows.sort(key=lambda r: r["ltv"], reverse=True)
134. summary = {
135. "run_date": TODAY,
136. "mode": cfg.get("mode"),
137. "armed": armed,
138. "not_armed_reasons": reasons,
139. "customers": len(metrics),
140. "baseline_as_of": h["meta"].get("baseline_as_of"),
141. "last_ingest": h["meta"].get("last_ingest"),
142. "tier_counts": tier_counts,
143. "stub_status_counts": status_counts,
144. }
145. out = export_spreadsheet(rows, summary)
147. # snapshot for the no-shrink check (keyed like the history store)
148. snap = {k: {"ltv": m["ltv"], "n": m["total_orders_count"]} for k, m in metrics.items()}
149. h["meta"].setdefault("computed_runs", []).append({"date": TODAY, "customers": len(metrics)})
150. h["meta"]["previous_computed"], h["meta"]["current_computed"] = (
151. h["meta"].get("current_computed", {}), snap)
152. save_history(h)
154. print(f"{'ARMED' if armed else 'READ-ONLY (thresholds not set)'}: "
155. f"{len(metrics)} customers from history "
156. f"(baseline as_of {h['meta'].get('baseline_as_of')}).")
157. if not armed:
158. print("Not armed:", "; ".join(reasons))
159. print("Spreadsheet:", out)
160. print("Plan JSON:", out.with_suffix("").with_suffix(".plan.json"))
163. def cmd_verify(args):
164. h = load_history()
165. prev = h["meta"].get("previous_computed") or {}
166. cur = h["meta"].get("current_computed") or {}
167. if not cur:
168. sys.exit("Nothing to verify: run classify first.")
169. shrunk = []
170. for key, now in cur.items():
171. before = prev.get(key)
172. if before and (now["ltv"] < before["ltv"] or now["n"] < before["n"]):
173. c = h["customers"].get(key, {})
174. shrunk.append({
175. "key": key, "email": c.get("email"), "company": c.get("company"),
176. "ltv_before": before["ltv"], "ltv_now": now["ltv"],
177. "orders_before": before["n"], "orders_now": now["n"],
178. })
179. report = OUT_DIR / f"verify_{TODAY}.json"
180. OUT_DIR.mkdir(parents=True, exist_ok=True)
181. report.write_text(json.dumps({"run": TODAY, "shrunk": shrunk}, indent=2), encoding="utf-8")
182. if shrunk:
183. print(f"NO-SHRINK GUARDRAIL: {len(shrunk)} customer(s) DECREASED: "
184. "lifetime history can only grow, so this means data went missing "
185. "(truncated pull, status reversal, or a bad ingest).")
186. for s in shrunk[:10]:
187. print(f" {s['email'] or s['key']}: ltv {s['ltv_before']} -> {s['ltv_now']}, "
188. f"orders {s['orders_before']} -> {s['orders_now']}")
189. print(f"Full report: {report}")
190. print("DO NOT demote tiers or move anyone to Winback off this run. "
191. "Flag for Matt/Hazel review.")
192. sys.exit(2)
193. print(f"Verify OK: no customer shrank vs the previous run. Report: {report}")