Matchms Spectral Matching

uv pip install matchms numpy pandas
# For chemical structure processing (SMILES, InChI, fingerprints):
uv pip install matchms[chemistry]

decisions:
  - id: D1
    param: referenceLibrary
    kind: required
    source: user
    ask: "Which spectral library should queries be matched against?"
    default: null

  - id: D2
    param: similarityMeasure
    kind: required
    source: user
    ask: "Should matching require the same precursor mass, or allow a mass shift so modified analogues still match?"
    default: "cosine on matched peaks, precursor mass fixed"

  - id: D3
    param: peakMatchTolerance
    kind: required
    source: data
    ask: "How close must two peaks be in m/z to count as the same fragment?"
    default: "0.1 Da - tighten toward 0.005 for high-resolution data"

  - id: D4
    param: spectrumPreprocessing
    kind: required
    source: user
    ask: "Which peaks should be discarded before scoring - low-intensity noise, peaks near the precursor, or spectra with too few peaks to be informative?"
    default: "drop peaks under 1% relative intensity, require at least 10 peaks"

  - id: D5
    param: scoreCutoff
    kind: required
    source: user
    ask: "What similarity score, and how many matched peaks, should a hit need before it is reported as an identification?"
    default: null

  - id: D6
    param: peakWeighting
    kind: optional
    source: user
    ask: "Should the score weight fragment m/z as well as intensity, favouring informative high-mass fragments?"
    default: "intensity only"

  - id: D7
    param: fingerprintSettings
    kind: optional_conditional
    source: user
    ask: "Which molecular fingerprint should structural similarity use, when comparing hits to known structures?"
    default: "not computed"

from matchms.importing import load_from_mgf
from matchms.filtering import default_filters, normalize_intensities
from matchms.filtering import select_by_relative_intensity, require_minimum_number_of_peaks
from matchms import calculate_scores
from matchms.similarity import CosineGreedy

# Load and process query spectra
queries = list(load_from_mgf("queries.mgf"))
queries = [default_filters(s) for s in queries]
queries = [normalize_intensities(s) for s in queries if s is not None]
queries = [require_minimum_number_of_peaks(s, n_required=5) for s in queries if s is not None]

# Load reference library
refs = list(load_from_mgf("library.mgf"))
refs = [default_filters(s) for s in refs]
refs = [normalize_intensities(s) for s in refs if s is not None]

# Calculate similarity scores
scores = calculate_scores(references=refs, queries=queries,
                          similarity_function=CosineGreedy(tolerance=0.1))
# Get best matches for first query
best = scores.scores_by_query(queries[0], sort=True)[:5]
for match, score_tuple in best:
    print(f"Score: {score_tuple['score']:.3f}, Matches: {score_tuple['matches']}")

from matchms.importing import (load_from_mgf, load_from_mzml, load_from_msp,
                                load_from_json, load_from_mzxml, load_from_pickle,
                                load_from_usi)
from matchms.exporting import save_as_mgf, save_as_msp, save_as_json, save_as_pickle

# Import from various formats (returns generators)
spectra_mgf = list(load_from_mgf("library.mgf"))
spectra_mzml = list(load_from_mzml("data.mzML"))
spectra_msp = list(load_from_msp("nist_library.msp"))
spectra_json = list(load_from_json("gnps_spectra.json"))
print(f"Loaded: MGF={len(spectra_mgf)}, mzML={len(spectra_mzml)}")

# Export processed spectra
save_as_mgf(spectra_mgf, "processed.mgf")
save_as_json(spectra_mgf, "processed.json")
save_as_pickle(spectra_mgf, "spectra.pickle")  # Fast for intermediate results

# Pickle for large datasets (fastest I/O)
from matchms.importing import load_from_pickle
spectra = list(load_from_pickle("spectra.pickle"))

from matchms import Spectrum
import numpy as np

# Create spectrum manually
mz = np.array([100.0, 150.0, 200.0, 250.0, 300.0])
intensities = np.array([0.1, 0.5, 0.9, 0.3, 0.7])
metadata = {
    "precursor_mz": 325.5,
    "ionmode": "positive",
    "compound_name": "Caffeine",
    "smiles": "CN1C=NC2=C1C(=O)N(C(=O)N2C)C"
}
spectrum = Spectrum(mz=mz, intensities=intensities, metadata=metadata)

# Access spectrum data
print(f"Peaks: {spectrum.peaks.mz}")
print(f"Precursor: {spectrum.get('precursor_mz')}")
print(f"Name: {spectrum.get('compound_name')}")

# Visualize
spectrum.plot()

from matchms.filtering import (
    default_filters, normalize_intensities,
    select_by_relative_intensity, select_by_mz,
    require_minimum_number_of_peaks, reduce_to_number_of_peaks,
    remove_peaks_around_precursor_mz, add_losses
)

# default_filters applies: metadata cleanup, charge correction, adduct parsing
spectrum = default_filters(spectrum)

# Peak normalization (max intensity → 1.0)
spectrum = normalize_intensities(spectrum)

# Filter peaks by relative intensity (remove noise below 1%)
spectrum = select_by_relative_intensity(spectrum, intensity_from=0.01, intensity_to=1.0)

# Filter peaks by m/z range
spectrum = select_by_mz(spectrum, mz_from=50.0, mz_to=500.0)

# Keep top N peaks only
spectrum = reduce_to_number_of_peaks(spectrum, n_max=50)

# Remove peaks near precursor (common contaminants)
spectrum = remove_peaks_around_precursor_mz(spectrum, mz_tolerance=17.0)

# Require minimum peaks for matching
spectrum = require_minimum_number_of_peaks(spectrum, n_required=5)

# Add neutral losses (useful for NeutralLossesCosine)
spectrum = add_losses(spectrum)

if spectrum is not None:
    print(f"After filtering: {len(spectrum.peaks.mz)} peaks")

# Chemical annotation filters (require matchms[chemistry])
from matchms.filtering import (
    derive_inchi_from_smiles, derive_inchikey_from_inchi,
    derive_smiles_from_inchi, add_fingerprint,
    repair_inchi_inchikey_smiles, require_valid_annotation
)

# Derive chemical identifiers from SMILES
spectrum = derive_inchi_from_smiles(spectrum)
spectrum = derive_inchikey_from_inchi(spectrum)

# Add molecular fingerprint for structural similarity
spectrum = add_fingerprint(spectrum, fingerprint_type="morgan", nbits=2048)

# Validate annotations
spectrum = require_valid_annotation(spectrum)
if spectrum is not None:
    print(f"InChIKey: {spectrum.get('inchikey')}")

from matchms import calculate_scores
from matchms.similarity import (
    CosineGreedy, CosineHungarian, ModifiedCosine,
    NeutralLossesCosine, FingerprintSimilarity,
    MetadataMatch, PrecursorMzMatch
)

# CosineGreedy — fast peak matching (greedy algorithm)
scores = calculate_scores(references=library, queries=unknowns,
                          similarity_function=CosineGreedy(tolerance=0.1))

# ModifiedCosine — accounts for precursor mass differences (best for analog search)
scores = calculate_scores(references=library, queries=unknowns,
                          similarity_function=ModifiedCosine(tolerance=0.1))

# CosineHungarian — optimal peak matching (slower but more accurate)
scores = calculate_scores(references=library, queries=unknowns,
                          similarity_function=CosineHungarian(tolerance=0.1))

# NeutralLossesCosine — similarity based on neutral loss patterns
scores = calculate_scores(references=library, queries=unknowns,
                          similarity_function=NeutralLossesCosine(tolerance=0.1))

# Access results
for i, query in enumerate(unknowns[:3]):
    best_matches = scores.scores_by_query(query, sort=True)[:3]
    print(f"\nQuery {i}: precursor_mz={query.get('precursor_mz')}")
    for ref, score_tuple in best_matches:
        print(f"  {ref.get('compound_name', 'Unknown')}: "
              f"score={score_tuple['score']:.3f}, matches={score_tuple['matches']}")

# FingerprintSimilarity — structural similarity (requires fingerprints)
from matchms.similarity import FingerprintSimilarity
scores = calculate_scores(references=library, queries=unknowns,
                          similarity_function=FingerprintSimilarity(
                              similarity_measure="jaccard"))

# PrecursorMzMatch — fast mass-based pre-filtering
from matchms.similarity import PrecursorMzMatch
scores = calculate_scores(references=library, queries=unknowns,
                          similarity_function=PrecursorMzMatch(tolerance=0.1))

# Multi-metric scoring: combine peak + structural similarity
cosine_scores = calculate_scores(references=library, queries=unknowns,
                                  similarity_function=CosineGreedy(tolerance=0.1))
fp_scores = calculate_scores(references=library, queries=unknowns,
                              similarity_function=FingerprintSimilarity())

from matchms import SpectrumProcessor
from matchms.filtering import (
    default_filters, normalize_intensities,
    select_by_relative_intensity, require_minimum_number_of_peaks,
    remove_peaks_around_precursor_mz, add_losses
)

# Define reusable pipeline
pipeline = SpectrumProcessor([
    default_filters,
    normalize_intensities,
    lambda s: select_by_relative_intensity(s, intensity_from=0.01),
    lambda s: remove_peaks_around_precursor_mz(s, mz_tolerance=17.0),
    lambda s: require_minimum_number_of_peaks(s, n_required=5),
    add_losses
])

# Apply to all spectra (filters returning None remove the spectrum)
processed = [pipeline(s) for s in raw_spectra]
processed = [s for s in processed if s is not None]
print(f"Processed: {len(processed)}/{len(raw_spectra)} spectra retained")

from matchms.importing import load_from_mgf
from matchms.filtering import default_filters, normalize_intensities
from matchms.filtering import select_by_relative_intensity, require_minimum_number_of_peaks
from matchms import calculate_scores
from matchms.similarity import ModifiedCosine
import pandas as pd

# Load and process both queries and library identically
def process_spectra(spectra):
    processed = []
    for s in spectra:
        s = default_filters(s)
        if s is None: continue
        s = normalize_intensities(s)
        s = select_by_relative_intensity(s, intensity_from=0.01)
        s = require_minimum_number_of_peaks(s, n_required=5)
        if s is not None:
            processed.append(s)
    return processed

queries = process_spectra(load_from_mgf("unknowns.mgf"))
library = process_spectra(load_from_mgf("reference_library.mgf"))
print(f"Queries: {len(queries)}, Library: {len(library)}")

# Score all query-reference pairs
scores = calculate_scores(references=library, queries=queries,
                          similarity_function=ModifiedCosine(tolerance=0.1))

# Extract best matches
results = []
for query in queries:
    best = scores.scores_by_query(query, sort=True)[:1]
    if best:
        ref, score_tuple = best[0]
        results.append({
            "query_precursor_mz": query.get("precursor_mz"),
            "match_name": ref.get("compound_name", "Unknown"),
            "match_smiles": ref.get("smiles", ""),
            "score": score_tuple["score"],
            "matched_peaks": score_tuple["matches"]
        })

df = pd.DataFrame(results)
confident = df[df["score"] > 0.7]
print(f"Confident matches (score>0.7): {len(confident)}/{len(df)}")
df.to_csv("identification_results.csv", index=False)

from matchms.importing import load_from_msp
from matchms.exporting import save_as_mgf
from matchms import SpectrumProcessor
from matchms.filtering import (
    default_filters, normalize_intensities,
    select_by_relative_intensity, require_minimum_number_of_peaks,
    require_precursor_mz, add_parent_mass
)

# Define QC pipeline
qc_pipeline = SpectrumProcessor([
    default_filters,
    require_precursor_mz,
    add_parent_mass,
    normalize_intensities,
    lambda s: select_by_relative_intensity(s, intensity_from=0.001),
    lambda s: require_minimum_number_of_peaks(s, n_required=3)
])

# Process and filter
raw = list(load_from_msp("raw_library.msp"))
cleaned = [qc_pipeline(s) for s in raw]
cleaned = [s for s in cleaned if s is not None]

print(f"Input: {len(raw)}, Output: {len(cleaned)} ({len(cleaned)/len(raw)*100:.1f}% retained)")
save_as_mgf(cleaned, "cleaned_library.mgf")