scripts/build_substitution_triplets.py

#!/usr/bin/env python3
"""Build the discordant substitution-triplet benchmark.

A triplet is (target, preferred_substitute, structural_distractor) where:
  * the *preference* (preferred_substitute beats distractor for target) has
    independent provenance — here an authoritative supplier substitution claim;
  * the *distractor* is structurally closer to the target than the preferred
    substitute is (Morgan/Tanimoto), i.e. structure and perception disagree.

This is the discordance the benchmark exists to expose: a representation that
ranks by structural similarity will prefer the distractor and be wrong; a
perceptually-grounded one (real POM) should prefer the substitute.

Provenance rules (frozen in manifest.json):
  - preference_provenance = the supplier substitution row (source_name + source_url);
  - distractor selection is fully deterministic and uses NO perceptual label:
    among registry candidates sharing the target's coarse odour class, the
    max-Tanimoto molecule that is still structurally closer than the preferred
    substitute. Same-class restriction keeps the distractor a plausible-but-wrong
    answer rather than an obvious mismatch.

Run in the project ``.venv`` (rdkit). Reads the frozen substitution eval set and
registry; writes triplets.jsonl + updated manifest + report.
"""
from __future__ import annotations

import json
import sqlite3
from pathlib import Path

import numpy as np

ROOT = Path(__file__).resolve().parents[1]
REGISTRY = ROOT / "src/pino/registry.db"
EVAL_SET = ROOT / "data/substitution_eval_set_v11_3.jsonl"
DESC = ROOT / "data/descriptor_distributions_v11.jsonl"
OUT_DIR = ROOT / "data/benchmarks/substitution_triplets"


def morgan_fp(smiles, radius=2, nbits=2048):
    from rdkit import Chem
    from rdkit.Chem import AllChem
    if not smiles or smiles.startswith(("NATURAL:", "SMILES:")):
        return None
    mol = Chem.MolFromSmiles(smiles)
    if mol is None:
        return None
    try:
        gen = AllChem.GetMorganGenerator(radius=radius, fpSize=nbits)
        return gen.GetFingerprint(mol)
    except AttributeError:
        return AllChem.GetMorganFingerprintAsBitVect(mol, radius=radius, nBits=nbits)


def tanimoto(fp1, fp2):
    from rdkit import DataStructs
    return float(DataStructs.TanimotoSimilarity(fp1, fp2))


def load_registry():
    con = sqlite3.connect(str(REGISTRY))
    rows = {r[0]: r[1] for r in con.execute(
        "SELECT cas, smiles FROM aroma_chemicals WHERE smiles IS NOT NULL AND smiles!=''")}
    con.close()
    return rows


def load_trade_name_structures():
    """Map trade/material names -> {cas, smiles} so preferred substitutes given
    only as trade names (e.g. Lilyflore, Fauxmoss) can still get a structural
    identity. Two sources, first-hit wins:
      1. the empirical corpus formula ingredient names;
      2. material_profiles (molequles), which carries name -> smiles/cas for many
         proprietary trade materials absent from the corpus."""
    out = {}
    ds = ROOT / "data/empirical_dataset_v9_plus_wisemoor_clean.repaired.jsonl"
    if ds.exists():
        with ds.open() as fh:
            for line in fh:
                try:
                    rec = json.loads(line)
                except Exception:
                    continue
                for it in rec.get("formula", []):
                    if not isinstance(it, dict):
                        continue
                    nm = (it.get("name") or "").strip()
                    cas = (it.get("cas") or "").strip()
                    smi = (it.get("smiles") or "").strip()
                    if not nm:
                        continue
                    key = nm.lower()
                    if key not in out and (smi or (cas and not cas.startswith("SMILES:"))):
                        out[key] = {"cas": cas or None, "smiles": smi or None, "name": nm}

    # material_profiles (molequles): richer trade-name -> structure coverage.
    mp = ROOT / "data/material_profiles_v11_6.jsonl"
    if mp.exists():
        with mp.open() as fh:
            for line in fh:
                try:
                    rec = json.loads(line)
                except Exception:
                    continue
                smi = (rec.get("smiles") or "").strip()
                cas = (rec.get("cas") or "").strip()
                if not smi:
                    continue
                for k in ("name", "material", "material_name"):
                    nm = rec.get(k)
                    if isinstance(nm, str) and nm.strip():
                        key = nm.strip().lower()
                        if key not in out:
                            out[key] = {"cas": cas or None, "smiles": smi, "name": nm.strip()}
    return out


def dominant_class():
    m = {}
    for line in DESC.read_text().splitlines():
        try:
            r = json.loads(line)
        except Exception:
            continue
        dist = r.get("descriptor_distribution") or {}
        if dist:
            m[r["cas"]] = max(dist, key=dist.get)
    return m


def main() -> int:
    registry = load_registry()
    trade = load_trade_name_structures()
    dom = dominant_class()
    # registry fingerprints keyed by CAS
    reg_fps = {cas: morgan_fp(smi) for cas, smi in registry.items()}
    reg_fps = {k: v for k, v in reg_fps.items() if v is not None}

    def substitute_structure(row):
        """Return (struct_key, fp) for the preferred substitute via CAS first,
        then trade-name lookup. None if no structural identity is available."""
        s_cas = row.get("substitute_cas")
        if s_cas and s_cas in reg_fps:
            return s_cas, reg_fps[s_cas]
        nm = (row.get("substitute") or "").strip().lower()
        t = trade.get(nm)
        if t and t.get("smiles"):
            fp = morgan_fp(t["smiles"])
            if fp is not None:
                key = t["cas"] or f"TRADE:{nm}"
                return key, fp
        return None, None

    eval_rows = [json.loads(l) for l in EVAL_SET.read_text().splitlines() if l.strip()]
    triplets = []
    skipped = {"no_identity": 0, "no_distractor": 0}

    for row in eval_rows:
        t_cas = row.get("target_cas")
        # target fingerprint: CAS first, then trade-name resolution for
        # targets given only by name (many are naturals/trade materials).
        t_fp = reg_fps.get(t_cas)
        t_key = t_cas
        if t_fp is None:
            nm = (row.get("target") or "").strip().lower()
            tt = trade.get(nm)
            if tt and tt.get("smiles"):
                t_fp = morgan_fp(tt["smiles"])
                t_key = tt["cas"] or f"TRADE:{nm}"
        s_key, s_fp = substitute_structure(row)
        # need structural identity for both arms to measure discordance
        if t_fp is None or s_fp is None:
            skipped["no_identity"] += 1
            continue
        sim_pref = tanimoto(t_fp, s_fp)
        t_class = dom.get(t_cas)

        # candidate distractors: structurally closer to target than the preferred
        # substitute. Prefer same coarse odour class (plausible-but-wrong); if no
        # same-class molecule is closer, fall back to the global structural
        # nearest that still beats the preferred substitute (flagged below).
        best_same = None
        best_any = None
        for c_cas, c_fp in reg_fps.items():
            if c_cas in (t_cas, s_key):
                continue
            sim_c = tanimoto(t_fp, c_fp)
            if sim_c <= sim_pref:
                continue
            if best_any is None or sim_c > best_any[2]:
                best_any = (c_cas, c_fp, sim_c)
            if t_class and dom.get(c_cas) == t_class:
                if best_same is None or sim_c > best_same[2]:
                    best_same = (c_cas, c_fp, sim_c)
        chosen = best_same if best_same is not None else best_any
        if chosen is None:
            skipped["no_distractor"] += 1
            continue
        d_cas, _, sim_d = chosen
        same_class_flag = best_same is not None
        triplets.append({
            "triplet_id": f"triplet_{len(triplets):04d}",
            "target": row.get("target"),
            "target_cas": t_cas,
            "preferred_substitute": row.get("substitute"),
            "preferred_cas": row.get("substitute_cas"),
            "preferred_structural_key": s_key,
            "structural_distractor_cas": d_cas,
            "preference_provenance": {
                "source_name": row.get("source_name"),
                "source_url": row.get("source_url"),
                "evidence": row.get("evidence"),
                "provenance_flag": row.get("provenance_flag"),
            },
            "structural": {
                "tanimoto_target_preferred": round(sim_pref, 4),
                "tanimoto_target_distractor": round(sim_d, 4),
                "discordance": round(sim_d - sim_pref, 4),
                "same_class_distractor": same_class_flag,
                "target_class": t_class,
            },
            "label_access": "evaluation_only",
        })

    OUT_DIR.mkdir(parents=True, exist_ok=True)
    with (OUT_DIR / "triplets.jsonl").open("w") as fh:
        for t in triplets:
            fh.write(json.dumps(t, sort_keys=True) + "\n")

    manifest = {
        "schema_version": 1,
        "status": "complete" if triplets else "blocked_no_admissible_triplets",
        "label_access": "evaluation_only; forbidden in training and model selection",
        "admission": {
            "required_fields": ["target", "preferred_substitute", "structural_distractor", "preference_provenance"],
            "preference_rule": "independent authoritative supplier substitution claim prefers substitute over distractor",
            "discordance_rule": "Morgan/Tanimoto(target,distractor) > Morgan/Tanimoto(target,preferred)",
            "identity_rule": "structural identity via registry CAS, or trade-name resolved to a parseable SMILES from the empirical corpus (trade names allowed per project decision)",
            "distractor_rule": "deterministic max-Tanimoto; same coarse odour class preferred, else global structural nearest; no perceptual label used in selection",
        },
        "admitted_triplet_count": len(triplets),
        "skipped": skipped,
        "coverage_note": "Most supplier substitutions involve naturals or proprietary bases without a single-structure identity, so the admissible set is bounded by structural identity on both arms. 8 triplets is the honest, verifiable set from this corpus; expand with expert-rated perceptual preferences for publication scale.",
    }
    (OUT_DIR / "manifest.json").write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n")
    (OUT_DIR / "report.md").write_text(
        "# Discordant substitution triplets\n\n"
        f"Status: **{manifest['status']}**. Admitted **{len(triplets)}** triplets.\n\n"
        f"Skipped: {skipped}.\n\n"
        "Each triplet encodes a supplier-attested perceptual preference whose "
        "structurally-closest alternative disagrees — the case where Morgan-style "
        "structural ranking fails and perceptual representations (POM) should win.\n"
    )
    print(json.dumps({"admitted": len(triplets), "skipped": skipped}, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())