pycsamt.agents.anomaly_agent#

pycsamt.agents.anomaly_agent#

AnomalyDetectionAgent — Unsupervised MT data anomaly detection.

Wraps AnomalyDetector:

A convolutional autoencoder (CAE) is trained on the clean portions of the dataset in an unsupervised manner. Observations whose reconstruction error exceeds the threshold_percentile are flagged as anomalies.

This agent detects anomalous (station, frequency) cells that classical rule-based QC may miss — e.g. coherent noise from a nearby source, subtle sensor drift, or 3-D scattering that deviates from the expected frequency dependence.

Requires PyTorch or TensorFlow.

Classes

AnomalyDetectionAgent(*[, api_key, model, ...])

Detect anomalous (station, frequency) observations in MT data.
class pycsamt.agents.anomaly_agent.AnomalyDetectionAgent(*, api_key=None, model=None, llm_provider='claude', threshold_percentile=95.0, latent_dim=32, epochs=50)[source]#

Bases: BaseAgent

Detect anomalous (station, frequency) observations in MT data.

Parameters:

  • api_key (str)

  • model (str)

  • llm_provider (str)

  • threshold_percentile (float) – Percentile of reconstruction errors used as the flagging threshold (default 95 — top 5 % are anomalies).

  • latent_dim (int) – CAE latent space dimension (default 32).

  • epochs (int) – Training epochs (default 50).

  • keys (Output data)

  • ----------

  • path (sites /)

  • output_dir (str, optional)

  • keys

  • ----------------

  • per-(station (anomaly_scores ndarray —)

  • error (freq) reconstruction)

  • anomalous (flags ndarray bool — True =)

  • {station (flag_table pandas DataFrame)

  • freq

  • score

  • flagged}

  • int (n_flagged)

  • list[str] (flagged_stations)

  • dict (figure_paths)

  • dict

SYSTEM_PROMPT: str = 'You are an expert in unsupervised anomaly detection for MT/AMT data.\nGiven an anomaly detection result, write 3-4 sentences that:\n1. State how many observations were flagged as anomalous and their distribution.\n2. Identify which stations or frequency bands are most affected.\n3. Diagnose the likely source (powerline harmonics, near-field, 3-D, instrument).\n4. Recommend whether to mask flagged data or apply targeted filtering.\nReply in plain English.\n'#

Override in subclasses to give the LLM its domain expertise.

execute(input_data)[source]#

Run this agent on input_data and return an AgentResult.

Subclasses must implement this method. The contract:

  • Reset self._last_cost = 0.0 at the top.

  • Record wall-clock time with t0 = time.time().

  • Return AgentResult(elapsed_seconds=time.time()-t0, cost_estimate_usd=self._last_cost, ...).

Parameters:

input_data (dict[str, Any])

Return type:

AgentResult