#!/usr/bin/env python """Template-generated synthetic supplement for phishing fine-tuning (TRAINING DATA ONLY). What it is A small (<= config data.synthetic.max_rows = 2,000), label-balanced set of short emails in the register a small-business IT consultancy's clients actually see. Positive rows (label 1) imitate the lure categories those clients get targeted with; negative rows (label 0) are ordinary business mail written in the same tone so the model cannot shortcut on register alone. Every row is fabricated from schematic templates plus seeded random slot filling -- no network, no LLM calls, no real messages. Organisations and people are placeholders ("Northwind Dental", a fixed fictional name list); every domain sits under the reserved .example / .invalid TLDs, IPs use the RFC 5737 documentation ranges and phone numbers use 555-01xx. Rows carry source="synthetic", doc_id "synthetic:NNNNNN", cluster_id -1, and the parquet schema carries {"generator": "make_synthetic.py", "note": "template-generated training supplement"}. The file is never a deliverable and must never be presented as real mail. How it is used and reported * finetune.py --synthetic on concatenates data/processed/synthetic.parquet onto finetune_train ONLY. It never enters finetune_val (model selection / early stopping stay on real data), the TREC-06 holdout, the tokenizer corpus or MLM pretraining. * Every headline metric is reported twice -- one run with and one without the supplement, identical dials otherwise -- and reports/experiments.jsonl records dials.synthetic so the comparison is auditable. The supplement is ~1.3% of the 157k training rows; its job is to nudge the model toward the business-email-compromise / Microsoft 365 lure register that the 2005-2008 spam corpora lack, not to move aggregate numbers. Any change on the holdout is reported as-is, including none. Shortcut control Templated data invites the classifier to learn the template rather than the phenomenon. The generator therefore (a) draws greetings, sign-offs, signatures, disclaimers, HTML wrappers, subject prefixes, casing, TLDs, URL schemes and the per-row length target from pools shared by both labels; (b) puts a link in BOTH classes at comparable rates: every negative category carries a {url} slot pointing at a plausible internal / vendor portal whose host prefixes (login., secure., docs., payments., *.sharepoint.*) and paths (/login, /auth/, /docs/, /inv/, ?id=) overlap the lookalike vocabulary used by the lures, so neither "contains a URL", "= 50 that is > 95% one class. The flagged list is written to _report.json ("token_purity_flagged") and logged, so a residual shortcut is auditable rather than hidden by a top-N cut-off. Exact duplicates are removed on common.normalize_text, so the written file's duplicate rate is 0. Three-way labels (label3 / label_set) The classifier now outputs Legitimate / Phishing / Spam. Every row carries, beside the binary `label`, a `label3` int64 (0 legitimate, 1 phishing; never 2 spam or -1 unwanted here) and a `label_set` string ("legit" / "phish"). The mapping is by category, i.e. by the "{rng.choice(anchors)}' return url def render(cat_name: str, ctx: dict[str, str], rng: random.Random) -> tuple[str, str]: cat = C[cat_name] html = rng.random() < cat.html_prob subject = _recase_subject(_fill(rng.choice(SUBJECT_PREFIX) + rng.choice(cat.subjects), ctx), rng) lines: list[str] = [] greet = _fill(rng.choice(GREETINGS), ctx) if greet: lines.append(greet) url = ctx["phish_url"] if cat.url == "phish" else ctx["legit_url"] body_sents: list[str] = [] for slot in cat.parts: choice = rng.choice(slot) if not choice: continue # weave the link in before slot filling so the anchor text (which may contain no braces) # survives str.format; {url} is the only marker not present in ctx. if "{url}" in choice: if cat.url == "none": continue choice = choice.replace("{url}", _anchor(url, rng, html, cat.anchors)) body_sents.append(_fill(choice, ctx)) lines.append(" ".join(body_sents)) signoff = _fill(rng.choice(SIGNOFFS), ctx) signature = _fill(rng.choice(SIGNATURES[rng.choice(cat.signers)]), ctx) disclaimer = _fill(rng.choice(DISCLAIMERS), ctx) if rng.random() < 0.3 else "" tail = [p for p in (signoff, signature, disclaimer) if p] # Per-row length target: right-skewed triangular over [MIN_WORDS, TARGET_MAX_WORDS] with the mode at # MIN_WORDS, so most rows stay short like real mail but a tail reaches ~200 words. The draw is # label-independent, and the padding comes from pools shared by both labels, so length and filler # vocabulary carry no label signal. The whole prospective message is counted so every body clears # MIN_WORDS. target = int(rng.triangular(MIN_WORDS, TARGET_MAX_WORDS, MIN_WORDS)) pad_pool = list(dict.fromkeys(cat.extras + SHARED_FILLER)) rng.shuffle(pad_pool) i = 0 tail_words = sum(len(p.split()) for p in tail) while len(" ".join(lines).split()) + tail_words < target and i < len(pad_pool): lines.append(_fill(pad_pool[i], ctx)) i += 1 lines.extend(tail) body = "\n\n".join(p for p in lines if p).strip() if html: body = "" + body.replace("\n\n", "

") + "" # trim to the upper word bound without cutting a word words = body.split() if len(words) > MAX_WORDS: body = " ".join(words[:MAX_WORDS]) return subject, body # ----------------------------------------------------------------------------- build def _token_purity(df: pd.DataFrame) -> dict[str, Any]: """Document-frequency label purity over the WHOLE vocabulary. For each lower-cased word token, count how many positive / negative documents contain it (presence, not multiplicity). Purity = max(pos, neg) / support. A high-support, high-purity token is a shortcut the classifier could exploit. The guard flags every token with support >= PURITY_SUPPORT and purity > PURITY_MAX regardless of rank; the top-N table is kept only for display, because a shortcut token confined to one category (support ~ n/16) never reaches the top of the frequency list. """ pos_df: Counter[str] = Counter() neg_df: Counter[str] = Counter() for text, label in zip(df["text"], df["label"]): toks = set(TOKEN_RE.findall(text.lower())) (pos_df if label == 1 else neg_df).update(toks) support = pos_df + neg_df def row(tok: str) -> dict[str, Any]: p, n = pos_df[tok], neg_df[tok] return {"token": tok, "support": support[tok], "pos": p, "neg": n, "purity": round(max(p, n) / support[tok], 3), "majority": 1 if p >= n else 0} top = [row(t) for t, _ in support.most_common(PURITY_TOP_N)] eligible = [t for t, s in support.items() if s >= PURITY_SUPPORT] flagged = sorted((row(t) for t in eligible if max(pos_df[t], neg_df[t]) / support[t] > PURITY_MAX), key=lambda r: (-r["support"], r["token"])) return {"top": top, "flagged": flagged, "n_tokens": len(support), "n_eligible": len(eligible)} def build(n: int, seed: int) -> tuple[pd.DataFrame, dict[str, Any]]: rng = random.Random(seed) half = n // 2 plan: list[str] = [] for i in range(half): plan.append(POSITIVE[i % len(POSITIVE)]) for i in range(n - half): plan.append(NEGATIVE[i % len(NEGATIVE)]) rng.shuffle(plan) seen: set[str] = set() records: list[dict[str, Any]] = [] attempts = 0 idx = 0 while idx < len(plan) and attempts < n * 60: attempts += 1 cat_name = plan[idx] ctx = make_ctx(rng) subject, body = render(cat_name, ctx, rng) text = f"{subject}\n\n{body}" key = common.normalize_text(text) if not key or key in seen: continue seen.add(key) records.append({ "doc_id": f"synthetic:{idx:06d}", "text": text, "subject": subject, "body": body, "label": C[cat_name].label, "source": "synthetic", "cluster_id": -1, "category": cat_name, }) idx += 1 if idx < len(plan): LOG.warning("only generated %d/%d unique rows before attempt cap", idx, len(plan)) # label3 / label_set are derived after generation so the RNG stream (and therefore every existing # column) is identical to the binary-era build for the same seed. for rec in records: rec["label3"] = label3_for_category(rec["category"]) rec["label_set"] = LABEL_SET_OF[rec["label3"]] df = pd.DataFrame.from_records(records, columns=COLUMNS) df["label3"] = df["label3"].astype("int64") assert ((df["label3"] != 0) == (df["label"] == 1)).all(), "label3 / binary label disagree" return df, _token_purity(df) def _length_stats(df: pd.DataFrame) -> dict[str, Any]: wl = df["body"].str.split().map(len) out: dict[str, Any] = {"min": int(wl.min()), "max": int(wl.max()), "mean": round(float(wl.mean()), 1), "median": int(wl.median())} for label in (1, 0): w = wl[df["label"] == label] out[f"label_{label}"] = {"min": int(w.min()), "max": int(w.max()), "mean": round(float(w.mean()), 1), "p10": int(w.quantile(0.1)), "p90": int(w.quantile(0.9))} return out def _url_stats(df: pd.DataFrame) -> dict[str, Any]: """Presence of a URL / HTML anchor / HTML wrapper per label; these must be close to balanced or the supplement teaches 'link => phishing'.""" out: dict[str, Any] = {} for key, pat in (("url", "://"), ("href", "")): hit = df["text"].str.contains(pat, regex=False) out[key] = {"pos": int(hit[df["label"] == 1].sum()), "neg": int(hit[df["label"] == 0].sum())} return out def _write_parquet(df: pd.DataFrame, out: Path) -> None: table = pa.Table.from_pandas(df, preserve_index=False) md = {**(table.schema.metadata or {}), **{k.encode(): v.encode() for k, v in PARQUET_META.items()}} table = table.replace_schema_metadata(md) out.parent.mkdir(parents=True, exist_ok=True) pq.write_table(table, out) def cmd_build(args: argparse.Namespace, cfg: dict) -> None: cap = int(cfg["data"]["synthetic"]["max_rows"]) n = args.n if args.n is not None else cap if n > cap: LOG.warning("requested n=%d exceeds config max_rows=%d; clamping", n, cap) n = cap out = Path(args.out) if args.out else Path(cfg["paths"]["data_processed"]) / "synthetic.parquet" if not out.is_absolute(): out = common.project_root() / out seed = args.seed if args.seed is not None else int(cfg["seed"]) df, purity = build(n, seed) _write_parquet(df, out) dup_rate = 1.0 - df["text"].map(common.normalize_text).nunique() / len(df) by_cat = df.groupby(["label", "category"]).size().reset_index(name="n") counts = { "rows": len(df), "label_1": int((df["label"] == 1).sum()), "label_0": int((df["label"] == 0).sum()), "per_category": {f"{int(r.label)}:{r.category}": int(r.n) for r in by_cat.itertuples()}, # Three-way counts (all three classes listed so a reader sees spam = 0 explicitly). "label3": {LABEL3_NAMES[k]: int((df["label3"] == k).sum()) for k in sorted(LABEL3_NAMES)}, "label_set": {v: int((df["label_set"] == v).sum()) for v in ("legit", "phish", "spam", "unwanted")}, "per_category_label3": {f"{int(r.label)}:{r.category}": label3_for_category(r.category) for r in by_cat.itertuples()}, } url_stats = _url_stats(df) report = { "generator": "make_synthetic.py", "note": PARQUET_META["note"], "seed": seed, "requested_n": n, "output": str(out), "duplicate_rate": round(float(dup_rate), 6), "counts": counts, "length_words": _length_stats(df), "url_presence": url_stats, "purity_support_threshold": PURITY_SUPPORT, "purity_warn_above": PURITY_MAX, "vocab_tokens": purity["n_tokens"], "tokens_with_support_ge_threshold": purity["n_eligible"], "token_purity_top30": purity["top"], "token_purity_flagged_count": len(purity["flagged"]), "token_purity_flagged": purity["flagged"], } report_path = out.with_name(out.stem + "_report.json") common.write_json(report, report_path) LOG.info("wrote %d rows (%d pos / %d neg) -> %s", counts["rows"], counts["label_1"], counts["label_0"], out) LOG.info("three-way: %s (label_set %s)", counts["label3"], counts["label_set"]) LOG.info("duplicate rate (normalize_text): %.6f", dup_rate) LOG.info("length words: %s", report["length_words"]) LOG.info("url presence (pos/neg): %s", url_stats) LOG.info("report -> %s", report_path) LOG.info("per-token label purity (top %d by support; support = #docs containing the token):", len(purity["top"])) LOG.info("%-16s%8s%7s%7s%8s", "token", "support", "pos", "neg", "purity") for r in purity["top"]: LOG.info("%-16s%8d%7d%7d%8.3f", r["token"], r["support"], r["pos"], r["neg"], r["purity"]) flagged = purity["flagged"] LOG.info("purity scan over full vocabulary: %d tokens, %d with support>=%d", purity["n_tokens"], purity["n_eligible"], PURITY_SUPPORT) if flagged: LOG.warning("%d token(s) with support>=%d are >%.0f%% one class (possible shortcut):", len(flagged), PURITY_SUPPORT, PURITY_MAX * 100) for r in flagged: LOG.warning(" %-16s support=%4d pos=%4d neg=%4d purity=%.3f majority=%d", r["token"], r["support"], r["pos"], r["neg"], r["purity"], r["majority"]) else: LOG.info("no token with support>=%d exceeds %.0f%% one class.", PURITY_SUPPORT, PURITY_MAX * 100) def cmd_show(args: argparse.Namespace, cfg: dict) -> None: path = Path(args.path) if args.path else Path(cfg["paths"]["data_processed"]) / "synthetic.parquet" if not path.is_absolute(): path = common.project_root() / path df = pd.read_parquet(path) for label in (1, 0): sub = df[df["label"] == label].head(args.k) print(f"\n{'=' * 78}\nlabel={label} ({'phishing/spam' if label else 'legitimate'}) - {len(sub)} samples") for row in sub.itertuples(): l3 = getattr(row, "label3", None) three = f" label3={l3} ({LABEL3_NAMES.get(l3, '?')}) label_set={getattr(row, 'label_set', '?')}" \ if l3 is not None else " label3=" print(f"\n--- {row.doc_id} [{row.category}] source={row.source} cluster_id={row.cluster_id}{three}") print(f"SUBJECT: {row.subject}") print(row.body) def main(argv: list[str] | None = None) -> None: ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--config", default="config.yaml") sub = ap.add_subparsers(dest="cmd", required=True) b = sub.add_parser("build", help="write the synthetic parquet + report json") b.add_argument("--n", type=int, default=None, help="rows to generate (default: config data.synthetic.max_rows)") b.add_argument("--out", default=None, help="output parquet path (default: data/processed/synthetic.parquet)") b.add_argument("--seed", type=int, default=None, help="override config seed") s = sub.add_parser("show", help="print k sample rows per class") s.add_argument("--k", type=int, default=6) s.add_argument("--path", default=None, help="parquet to read (default: data/processed/synthetic.parquet)") args = ap.parse_args(argv) cfg = common.load_config(args.config) global LOG LOG = common.setup_logging("make_synthetic", cfg["paths"]["logs"]) {"build": cmd_build, "show": cmd_show}[args.cmd](args, cfg) if __name__ == "__main__": main()