c3-charm

A Python SDK for the CHARM time-series foundation model. Provides embeddings (multivariate time series → vectors), forecast/backcast (quantile predictions), and a toolkit for downstream tasks (anomaly detection, retrieval, classification, reconstruction, forecasting).

What is CHARM?

CHARM (CHannel Aware Representation Model) is a foundation model for multivariate time series. It ingests windows of (T, C) data — T timesteps, C channels — and produces dense embeddings that capture temporal patterns and cross-channel relationships. Channel names (descriptions) are part of the input, making the model channel-aware.

No scaling required — the model handles normalization internally. Send raw data directly.

Dual-model serving. A CHARM server can be backed by two independent checkpoints: one for embeddings (/predict, exposed as client.embeddings) and a separate one for forecasting (/forecast, exposed as client.prediction). These may differ in architecture, patch size, and embedding dimension, so treat per-model properties as model-dependent and read them from client.model_info() rather than hardcoding.

Installation

pip install c3-charm            # core SDK only (embeddings + forecast)
pip install c3-charm[toolkit]   # includes PyTorch models, datasets, trainers

Or from source:

git clone https://github.com/c3ai/c3-charm.git
cd c3-charm
poetry install                    # core SDK only
poetry install --with toolkit     # include toolkit dependencies

CPU vs CUDA torch

pip install c3-charm pulls the default torch wheel from PyPI, which on Linux is the full CUDA build (~4 GB). The SDK itself never requires a GPU — the model runs server-side, and toolkit training heads are small enough to fit on CPU — so if you want the smaller CPU-only wheel, install torch from the CPU index before c3-charm:

pip install torch --index-url https://download.pytorch.org/whl/cpu
pip install c3-charm

Or from source, use bash setup.sh, which forces the CPU wheel.

If you instead want a specific CUDA build, install it manually after c3-charm:

pip install --force-reinstall --no-deps torch \
    --index-url https://download.pytorch.org/whl/cu124

Swap cu124 for the CUDA version that matches your driver (cu121, cu126, cu128, …). --no-deps is important — it stops pip from re-resolving your torch install.

Core SDK

Client initialization

from charm import CharmClient

client = CharmClient(
    base_url="http://your-server:8080",
    api_key="your-api-key",      # or set CHARM_API_KEY env var
    timeout=300,                 # override; SDK default is 15s — raise it for /forecast (server allows up to ~220s)
    max_retries=3,
)

Embeddings — client.embeddings.create()

Converts time series windows into dense vectors.

response = client.embeddings.create(
    descriptions=[["sensor_A", "sensor_B"]],  # (N, C) channel names
    ts_array=[[[1.0, 2.0], [1.1, 2.1], ...]],  # (N, T, C) values
    batch_size=32,
    return_tensors="np",       # "list", "np", or "torch"
    aggregate=True,            # True → (N, D); False → (N, T_, C, D)
    progress=True,
)
embeddings = response.embeds  # shape (N, D) when aggregate=True

aggregate parameter:

• True (default): Returns flattened embeddings (N, D) — one vector per series. Best for retrieval, classification, clustering.
• False: Returns per-patch, per-channel embeddings (N, T_, C, D) where T_ = ceil(T / patch_size). Best for fine-grained tasks or custom heads.

Async (faster for large datasets):

response = await client.embeddings.async_create(
    descriptions=descriptions,
    ts_array=ts_array,
    max_B_per_request=32,
    concurrency_per_call=8,
    return_tensors="np",
    aggregate=True,
)

Forecast / Backcast — client.prediction.create()

Zero-shot quantile predictions — no training required.

response = client.prediction.create(
    descriptions=[["sensor_A", "sensor_B"]],
    ts_array=[[[1.0, 2.0], [1.1, 2.1], ...]],
    target_len=10,       # positive = forecast, negative = backcast
    return_tensors="np",
)
forecast = response.denormalized_predictions  # (N, target_len=10, C, Q) — Q quantiles (model-dependent, e.g. 21)
median = response.median                      # (N, 10, C) — point forecast (median quantile)

Backcast (reconstruct past values):

response = client.prediction.create(
    descriptions=descriptions,
    ts_array=ts_array,
    target_len=-8,  # reconstruct last 8 steps
    return_tensors="np",
)

Input constraints

The model was trained on windows of up to 8192 timesteps (patch size 16). Inputs are not required to be a multiple of the patch size — they are padded to a patch boundary internally. The architecture can technically accept more than 8192 (up to 1500 patches), but longer inputs rely on positions seen only in pretraining, so quality degrades. Query client.model_info() for the served model's actual patch size and embedding dimension.

The SDK handles client-side batching automatically when you set batch_size (sync) or max_B_per_request (async).

Output shapes

Channel descriptions

Descriptions are required and affect embedding quality. They tell the model what each channel represents.

Good descriptions — use meaningful, consistent names:

descriptions = [["engine_temperature", "oil_pressure", "rpm"]]

Acceptable — short but informative:

descriptions = [["temp", "pressure", "speed"]]

Avoid — generic or positional names reduce model effectiveness:

descriptions = [["col_0", "col_1", "col_2"]]  # works but suboptimal

When working with pandas DataFrames, use column names directly:

descriptions = [df.columns.tolist()] * N

Scaling

No pre-processing needed. CHARM normalizes internally. Send raw data as-is. Do not apply StandardScaler, MinMaxScaler, or log transforms before calling the API.

Error handling

from charm import CharmError, AuthenticationError, InvalidRequestError, RateLimitError

try:
    response = client.embeddings.create(...)
except AuthenticationError:
    # bad API key
except InvalidRequestError as e:
    # shape violations, empty input
except RateLimitError:
    # back off and retry
except CharmError as e:
    # catch-all for other SDK errors

Toolkit — Downstream Tasks

The toolkit (pip install c3-charm[toolkit]) provides PyTorch models, dataset utilities, and training infrastructure for fine-tuning on top of CHARM embeddings.

Retrieval — charm_toolkit.retrieval

Find similar time series by embedding similarity.

from charm_toolkit.retrieval import (
    l2_normalize,
    cosine_similarity_matrix,
    knn_search,
    retrieval_metrics,
)

# Embed your data
response = client.embeddings.create(
    descriptions=descriptions,
    ts_array=windows_list,
    return_tensors="np",
)
embeddings = response.embeds  # (N, D)

# Similarity search
sim = cosine_similarity_matrix(embeddings, embeddings)

# kNN search
indices, scores = knn_search(query_emb, corpus_emb, k=5)

# Evaluation metrics
metrics = retrieval_metrics(
    query_emb=query_emb,
    corpus_emb=corpus_emb,
    query_labels=query_labels,
    corpus_labels=corpus_labels,
    k_values=[1, 3, 5, 10],
    exclude_self=True,
    query_ids=query_dataset_names,
    corpus_ids=corpus_dataset_names,
)
# Returns: precision@k, ndcg@k, hit_rate@k

Anomaly Detection — charm_toolkit.anomaly_detection

Detect anomalies via kNN distance scoring on windowed CHARM embeddings.

from charm_toolkit.anomaly_detection import (
    sliding_window_embeddings,
    knn_anomaly_scores,
    window_scores_to_pointwise,
)

# 1. Embed sliding windows
train_emb = sliding_window_embeddings(
    client, train_data, descriptions,
    window_size=128, stride=1, batch_size=64,
)
test_emb = sliding_window_embeddings(
    client, test_data, descriptions,
    window_size=128, stride=1, batch_size=64,
)

# 2. Score test windows by distance to train
window_scores = knn_anomaly_scores(
    test_emb=test_emb,
    reference_emb=train_emb,
    k=5,
    distance="cosine",    # "cosine", "l2", "l1"
    aggregation="mean",   # "mean", "max"
)

# 3. Aggregate to per-timestep scores
pointwise_scores = window_scores_to_pointwise(
    window_scores=window_scores,
    window_size=128,
    stride=1,
    total_length=len(test_data),
    method="mean",  # "mean", "max", "last", "center"
)

Pointwise aggregation methods:

Each timestep is covered by multiple overlapping windows. The method parameter controls how to assign a single score per timestep:

Zero-shot recipes (no clean reference set required) — recommended methods, in order of strength:

1. Bootstrap k-NN (best). Two steps: run sklearn.ensemble.IsolationForest on the embedding matrix and take the bottom ~70% by score as a presumed-clean reference; then call knn_anomaly_scores against that reference.
2. CBLOF: pyod.models.cblof.CBLOF on the embedding matrix.
3. IsolationForest: sklearn.ensemble.IsolationForest directly on the embedding matrix.

L2-normalize embeddings beforehand to use cosine geometry. See demo_retrieval_anomaly_detection.ipynb.

Best practices (validated on TSB-AD, metric VUS-PR):

1. Ensemble the embedding score with per-window statistics. The encoder instance-normalizes each window, erasing amplitude/level-shift anomalies (spikes, steps — the most common kind) from the embedding. Recover them with window_statistics (no extra model call) and combine via ensemble_scores (method="zscore" — z-score each detector and sum; parameter-free and as strong as a tuned weighted ensemble). This is worth ~+7 pp VUS-PR supervised.

   from charm_toolkit.anomaly_detection import window_statistics, ensemble_scores
   
   emb_score = knn_anomaly_scores(test_emb, train_emb, k=3, distance="cosine")
   S_tr = window_statistics(train_data, 128, 1)
   mu, sd = S_tr.mean(0, keepdims=True), S_tr.std(0, keepdims=True) + 1e-8
   stats_score = knn_anomaly_scores(
       (window_statistics(test_data, 128, 1) - mu) / sd, (S_tr - mu) / sd,
       k=3, distance="l2")
   window_scores = ensemble_scores([emb_score, stats_score], method="zscore")
   # zero-shot: build the reference by bootstrap (above) and use method="rank"
   

2. Multivariate: do NOT pool channels when the channel count is high. Channel-mean pooling dilutes an anomaly confined to a few channels across all of them. Keep channels separate with sliding_window_channel_embeddings + per_channel_knn_scores (channel_pool="adaptive"). The benefit grows with the number of channels — negligible at C≤3, sizeable at C≥20, large at C>60 — so it is advisable whenever C is high. For few channels, plain mean-pooling is equally good and cheaper.

   from charm_toolkit.anomaly_detection import (
       sliding_window_channel_embeddings, per_channel_knn_scores)
   
   train_pc = sliding_window_channel_embeddings(client, train_data, descriptions)  # (N, C, D)
   test_pc  = sliding_window_channel_embeddings(client, test_data, descriptions)
   emb_score = per_channel_knn_scores(test_pc, train_pc, k=3, channel_pool="adaptive")
   

3. No compute constraints? Use multiscale. Run the embedding detector at several window sizes (e.g. {64, 128, 256}) and the statistics detector at several more (e.g. {16, 32, 64, 128, 256, 512}), combine each family with multiscale_max_score, then sum the two combined results. This is the highest-accuracy recipe found (TSB-AD full eval, stride-1: VUS-PR 0.634 vs 0.607 for the single-scale ensemble above, +12.7 pp over the naive last-layer-mean baseline) at the cost of ~3-6x the detector calls per series (one embedding/statistics pass per window size). At this point the earlier channel-pooling choice stops mattering: plain mean-pooling ties or beats the per-channel/ adaptive options once multiscale statistics is already catching scale-localized anomalies, so no channel-count branching is needed at all.

   from charm_toolkit.anomaly_detection import multiscale_max_score
   
   def multiscale_detector(client, train, test, descriptions, scales, is_stats):
       scores = []
       for ws in scales:
           train_w = sliding_window_embeddings(client, train, descriptions, window_size=ws) if not is_stats \
               else window_statistics(train, window_size=ws)
           test_w = sliding_window_embeddings(client, test, descriptions, window_size=ws) if not is_stats \
               else window_statistics(test, window_size=ws)
           s = knn_anomaly_scores(test_w, train_w, k=3, distance=("cosine" if not is_stats else "l2"))
           scores.append(window_scores_to_pointwise(s, ws, 1, len(test)))
       return multiscale_max_score(scores)
   
   final_score = multiscale_detector(client, train, test, descriptions, [64, 128, 256], is_stats=False) \
       + multiscale_detector(client, train, test, descriptions, [16, 32, 64, 128, 256, 512], is_stats=True)
   

ReconstructionModel — anomaly detection via learned head

from charm_toolkit import (
    ReconstructionModel, create_reconstruction_datasets,
    collator, TrainerClass,
)
from torch.utils.data import DataLoader
import torch.nn as nn

train_ds, val_ds, test_ds = create_reconstruction_datasets(
    raw_data,           # (T, C) numpy array or torch tensor
    descriptions=channel_names,
    window_size=256,
    stride=1,
    train_ratio=0.7,
    val_ratio=0.15,
    sequential=True,
    scale=True,
)

model = ReconstructionModel(
    embedding_client=client,
    reconstructor="linear",  # "linear", "mlp", or custom nn.Module
    hidden_dim=128,
    dropout=0.1,
)

trainer = TrainerClass(
    model=model,
    train_loader=DataLoader(train_ds, batch_size=512, collate_fn=collator),
    val_loader=DataLoader(val_ds, batch_size=512, collate_fn=collator),
    epochs=1000,
    patience=5,
    lr=1e-3,
    criterion=nn.HuberLoss(),
)
trainer.fit()

ForecastingModel — embedding-based forecasting

from charm_toolkit import ForecastingModel, create_forecasting_datasets, collator, TrainerClass
from torch.utils.data import DataLoader

train_ds, val_ds, test_ds = create_forecasting_datasets(
    raw_data,
    descriptions=channel_names,
    train_horizon=96,
    test_horizon=96,
    train_ratio=0.7,
    val_ratio=0.15,
    sequential=True,
    scale=True,
)

model = ForecastingModel(
    embedding_client=client,
    horizon=96,
    input_size=96,
    head="linear",
    hidden_dim=128,
    mode="last",         # "last", "avg", "none"
    per_channel=True,
    num_channels=len(channel_names),
)

trainer = TrainerClass(
    model=model,
    train_loader=DataLoader(train_ds, batch_size=512, collate_fn=collator),
    val_loader=DataLoader(val_ds, batch_size=512, collate_fn=collator),
    epochs=1000,
    patience=10,
    lr=1e-2,
)
trainer.fit()

ClassificationModel — time series classification

from charm_toolkit import ClassificationModel, create_classification_datasets, collator, TrainerClass
from torch.utils.data import DataLoader
import torch.nn as nn

train_ds, val_ds, test_ds = create_classification_datasets(
    raw_data,          # (N, T, C)
    labels=labels,     # list of N integer labels
    descriptions=channel_names,
    train_ratio=0.7,
    val_ratio=0.15,
)

model = ClassificationModel(
    embedding_client=client,
    num_classes=num_classes,
    hidden_dim=128,
    pooling_over_t="mean",
    pooling_over_channels="mean",
    classifier_type="mlp",
)

trainer = TrainerClass(
    model=model,
    train_loader=DataLoader(train_ds, batch_size=32, collate_fn=collator),
    val_loader=DataLoader(val_ds, batch_size=32, collate_fn=collator),
    epochs=100,
    patience=10,
    lr=1e-3,
    criterion=nn.CrossEntropyLoss(),
)
trainer.fit()

Precomputing embeddings (critical for training)

Toolkit models call the API every forward pass. For training with hundreds of windows per epoch, precompute embeddings once:

from charm_toolkit import precompute_dataset_embeddings, PrecomputedEmbeddingsDataset

# Compute once, save to disk as memmap
train_shape = precompute_dataset_embeddings(
    client=client, dataset=train_ds,
    output_path="./outputs/train_embeddings.pt", memory_batch_size=8192
)
val_shape = precompute_dataset_embeddings(
    client=client, dataset=val_ds,
    output_path="./outputs/val_embeddings.pt", memory_batch_size=8192
)

# Wrap datasets — model skips API calls when "embeds" key present
train_ds = PrecomputedEmbeddingsDataset(train_ds, "./outputs/train_embeddings.pt", train_shape)
val_ds = PrecomputedEmbeddingsDataset(val_ds, "./outputs/val_embeddings.pt", val_shape)

# Training now uses cached embeddings — orders of magnitude faster
train_loader = DataLoader(train_ds, batch_size=512, shuffle=True, collate_fn=collator)

Trainer API

from charm_toolkit import TrainerClass

trainer = TrainerClass(
    model=model,
    train_loader=train_loader,
    val_loader=val_loader,
    test_loader=test_loader,     # optional
    lr=1e-3,
    weight_decay=1e-4,
    epochs=1000,
    patience=5,
    min_delta=1e-4,
    max_grad_norm=5.0,
    criterion=None,              # defaults to MSELoss
)
trainer.fit()
test_loss = trainer.evaluate(test_loader)

Dataset factory functions

All return (train_dataset, val_dataset, test_dataset):

Reconstruction and forecasting expect a single long time series (T, C) split temporally. Classification expects pre-windowed (N, T, C).

collator

All DataLoaders using toolkit datasets require collator as the collate_fn:

from charm_toolkit import collator
# or equivalently:
from charm_toolkit.Datasets import collator

Embeddings as features

CHARM embeddings work as drop-in feature vectors for any sklearn model:

import numpy as np
from sklearn.ensemble import IsolationForest
from sklearn.linear_model import LogisticRegression
from charm_toolkit.retrieval import cosine_similarity_matrix

response = client.embeddings.create(
    descriptions=descriptions,
    ts_array=windows_list,
    return_tensors="np",
)
X = response.embeds  # (N, D)

# Anomaly detection with isolation forest
clf = IsolationForest(contamination=0.05)
anomaly_labels = clf.fit_predict(X)

# Similarity search
sim = cosine_similarity_matrix(X, X)

# As features for any classifier
clf = LogisticRegression().fit(X_train, y_train)

Local Deployment

Deploy models locally from GitHub releases — no remote server needed:

with CharmClient(tag="experiment-2026-03-15_10-30-00") as client:
    response = client.embeddings.create(...)
# Server shuts down automatically

When tag is provided:

1. Checks for GPU availability (falls back to CPU)
2. Clones repo at the specified tag (shallow clone)
3. Downloads model weights from the GitHub release
4. Launches the serving stack locally
5. Polls health endpoint until ready

Files cached at ~/.charm/models/<tag>/ for fast subsequent runs.

CharmClient(
    tag="experiment-tag",           # required for local mode
    repo_url="https://...",         # default: c3-e/research
    cache_dir="/path/to/cache",     # default: ~/.charm/models
    port=8080,                      # 0 = auto-select
)

Best practices

Input & preprocessing

• Send data as (N, T, C) — N series, each T timesteps × C channels; all series in one request must share the same T and C.
• Keep T ≤ 8192 (the training window). The model can accept more, but quality degrades on lengths it wasn't trained on. T does not need to be a multiple of the patch size — inputs are padded to a patch boundary internally.
• Do not pre-scale your data. CHARM normalizes internally (asinh z-score); applying StandardScaler/MinMaxScaler/log yourself hurts results.

Channel descriptions (a real quality lever)

• Use meaningful, consistent channel names ("engine_temperature", not "col_0") — the model is channel-aware and descriptions materially affect embeddings.
• Reuse the same names across requests so embeddings stay comparable (retrieval, clustering).

Batching & throughput

• Respect the per-request budget: batch_size × C × T ≤ 500,000 (SDK-enforced client-side).
• Use async_create for large N — it batches concurrently; the sync client is sequential and slow past ~100 series. Tune max_B_per_request / concurrency_per_call instead of one giant request.

Timeouts & retries

• Raise timeout for forecasting — the SDK default is 15s, but the server allows forecasts up to ~220s. Use timeout ≈ 220+ for prediction.create.
• Keep the built-in retries (exponential backoff on 429/5xx) rather than hand-rolling.

Embeddings

• aggregate=True (default) → (N, D) for retrieval / classification / clustering. aggregate=False → (N, T_, C, D) only when you need per-patch/per-channel detail for a custom head.
• L2-normalize before cosine similarity (embeddings are unit-normed by the encoder, but normalize again after any pooling you do).
• Discover D at runtime via client.model_info() — it's model-dependent; don't hardcode.

Forecasting

• target_len > 0 = forecast, < 0 = backcast; 0 is invalid.
• Keep T + abs(target_len) ≤ 8192 (the training window) for best forecast/backcast quality.
• Use response.median for a point forecast, or the full quantile axis for intervals. Q is model-dependent (e.g. 21 or 99) — read denormalized_predictions.shape[-1].

Classification

• For classification heads, don't pool over channels — keeping the per-channel embeddings flat (rather than averaging them) boosts accuracy, at a modest cost in head size/complexity. In the toolkit ClassificationModel, set pooling_over_channels="flatten" (and pass num_channels=C, required for flatten) instead of the default "mean".

Dual-model awareness

• Treat /predict (embeddings) and /forecast as separate models — they may have different patch sizes / embedding dims. Read the per-role models map from client.model_info() instead of assuming they match.

Training on top of CHARM

• Precompute embeddings once (precompute_dataset_embeddings + PrecomputedEmbeddingsDataset) — toolkit models otherwise call the API every forward pass.

Reliability & ops

• Catch specific errors (InvalidRequestError, AuthenticationError, RateLimitError, or the base CharmError) and back off on rate limits.
• Use the context manager (with CharmClient(...) as client:) so local deployments shut down cleanly.
• Set credentials via env (CHARM_API_KEY, CHARM_BASE_URL) rather than hardcoding.

Decision guide

When to use CHARM

• Multivariate time series (multiple channels measured over time)
• Each window has at least a few patches (patch size is 16, so ~48+ timesteps is a good floor)
• You want a strong starting point without feature engineering

When to use classical methods instead

• Tabular data without a time dimension — use LightGBM, XGBoost
• Very short series (< 10 timesteps)
• Single scalar features — still works but may not outperform ARIMA/ETS

Zero-shot vs fine-tuned

Testing

pip install pytest
python -m pytest tests/
python -m pytest tests/test_utils.py -v

Documentation

The full API reference and usage guide is this README — it renders on the PyPI page.

License

Apache License 2.0 — see LICENSE.