alphainfo 1.4.1

Python client for alphainfo.io — Structure-aware analysis for any time series

alphainfo

Python client for the alphainfo Structural Intelligence API.

Detect structural regime changes in time series — biomedical signals, financial markets, energy grids, seismic data, IoT sensors, and more. No model training required.

from alphainfo import AlphaInfo

client = AlphaInfo(api_key="ai_your_key")
result = client.analyze(signal=ecg_data, sampling_rate=360.0, domain="biomedical")

print(result.confidence_band)   # 'stable', 'transition', or 'unstable'
print(result.structural_score)  # 0.0 to 1.0
print(result.analysis_id)       # UUID for audit trail

Installation

pip install alphainfo

# Optional: enable HTTP/2 for better connection efficiency
pip install alphainfo[http2]

Requires Python 3.8+. Core dependency: httpx.

Quick Start

1. Get your API key

Sign up at alphainfo.io/register — free tier includes 50 analyses/month.

2. Analyze a signal

from alphainfo import AlphaInfo

client = AlphaInfo(api_key="ai_your_key")

# Any time series: ECG, market prices, sensor readings, power grid...
result = client.analyze(
    signal=[1.2, 1.3, 1.1, 2.8, 3.1, 3.0, ...],
    sampling_rate=250.0,
    domain="biomedical",
)

if result.change_detected:
    print(f"Regime change detected! Band: {result.confidence_band}")
    print(f"Structural score: {result.structural_score:.3f}")
    print(f"Audit ID: {result.analysis_id}")

3. Analyze market data

# The API fetches market data automatically
market = client.analyze_market("AAPL", interval="1d")
print(f"AAPL regime: {market.confidence_band}")
print(f"Score: {market.structural_score:.3f}")

4. Batch analysis

# Analyze up to 100 signals in one call
batch = client.analyze_batch(
    signals=[signal_1, signal_2, signal_3],
    sampling_rate=1000.0,
    domain="sensors",
)

for item in batch.results:
    if item.success:
        print(f"Signal {item.index}: {item.confidence_band} ({item.structural_score:.3f})")
    else:
        print(f"Signal {item.index}: error — {item.error}")

5. Semantic layer (severity, trend, alerts)

result = client.analyze(
    signal=data, sampling_rate=1.0,
    include_semantic=True,
    baseline=calm_period,
)

if result.semantic:
    print(result.semantic.alert_level)       # 'normal', 'attention', 'alert', 'critical'
    print(result.semantic.severity)          # 'none', 'low', 'moderate', 'high', 'critical'
    print(result.semantic.severity_score)    # 0-100 (higher = more severe)
    print(result.semantic.trend)             # 'stable', 'diverging', 'monitoring'
    print(result.semantic.summary)           # "⚠️ Structural divergence detected (severity: high)"
    print(result.semantic.recommended_action)  # 'log_only', 'monitor', 'human_review', 'immediate_human_review'

# Short signal warning (< 100 samples)
if result.warning:
    print(result.warning)  # "Signal has only 30 samples..."

Severity thresholds:

severity	severity_score	Meaning
none	0-15	No structural degradation
low	16-35	Minor deviation, monitor
moderate	36-65	Notable change, investigate
high	66-85	Significant regime shift
critical	86-100	Severe structural breakdown

6. Multi-channel (vector) analysis with per-channel baselines

# Multi-lead ECG, multi-axis accelerometer, cross-asset finance...
vector = client.analyze_vector(
    channels={
        "lead_I": ecg_lead_1,
        "lead_II": ecg_lead_2,
        "lead_III": ecg_lead_3,
    },
    sampling_rate=360.0,
    domain="biomedical",
)

print(f"Aggregated score: {vector.structural_score:.3f}")
print(f"Composite band: {vector.confidence_band}")
for name, ch in vector.channels.items():
    print(f"  {name}: {ch.confidence_band} (score={ch.structural_score:.3f})")

# With per-channel baselines (e.g. calm period reference)
vector = client.analyze_vector(
    channels={"SPY": spy_data, "VIX": vix_data, "GLD": gld_data},
    sampling_rate=1.0,
    baselines={"SPY": spy_calm, "VIX": vix_calm, "GLD": gld_calm},
)

7. Audit trail

# Replay any past analysis
replay = client.audit_replay("550e8400-e29b-41d4-a716-446655440000")
print(f"Original score: {replay.output['structural_score']}")

# List recent analyses
history = client.audit_list(limit=10)
for entry in history:
    print(f"{entry.analysis_id} — {entry.structural_score}")

Async Support

from alphainfo import AsyncAlphaInfo

async with AsyncAlphaInfo(api_key="ai_your_key") as client:
    result = await client.analyze(signal=data, sampling_rate=250.0)
    market = await client.analyze_market("BTC-USD")

All methods available on AlphaInfo are also available on AsyncAlphaInfo.

Error Handling

from alphainfo import AlphaInfo, AuthError, RateLimitError, ValidationError

client = AlphaInfo(api_key="ai_your_key")

try:
    result = client.analyze(signal=data, sampling_rate=250.0)
except AuthError:
    print("Invalid API key")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after}s")
except ValidationError as e:
    print(f"Invalid input: {e.message}")

Exception hierarchy:

Exception	HTTP Code	When
AuthError	401	Invalid or missing API key
ValidationError	400, 413	Bad input or signal too large
RateLimitError	429	Quota or concurrency limit exceeded
NotFoundError	404	Analysis ID not found (audit)
APIError	5xx	Server error
TimeoutError	—	Request timed out after retries
NetworkError	—	Connection failed

All inherit from AlphaInfoError.

Configuration

client = AlphaInfo(
    api_key="ai_your_key",
    base_url="https://alphainfo.io",  # default
    timeout=30.0,                      # seconds (default)
    max_retries=3,                     # automatic retry on transient errors
    retry_base_delay=1.0,              # initial backoff delay (seconds)
    retry_max_delay=32.0,              # max delay between retries (seconds)
    http2=None,                        # auto-detect (True if h2 installed)
)

The client automatically retries on:

• Network timeouts and connection errors
• HTTP 429 (rate limits) — respects Retry-After header
• HTTP 5xx (server errors)

Non-retryable errors (401, 400, 404) are raised immediately.

Backoff is exponential: retry_base_delay * 2^attempt, capped at retry_max_delay.

Rate Limit Info

result = client.analyze(signal=data, sampling_rate=250.0)
info = client.rate_limit_info
if info:
    print(f"Remaining: {info.remaining}/{info.limit}")

Signal Size Guide

Samples	Behavior	Recommendation
< 10	Rejected (422)	Hard minimum
10-49	Returns 0.5 + warning	Too short for multiscale
50-99	Returns 0.5 + warning	Limited confidence
100-199	Variable scores	Detection active, less reliable
200-500	Reliable scores	Recommended range
500+	Reliable, may dilute point events	Use windowing for point detection

Note: sampling_rate controls multiscale window sizing but does not change scores for a given signal. For daily financial data use sampling_rate=1.0; for ECG at 250Hz use sampling_rate=250.0.

Domains

Domain	Use case
generic	Default — works for any signal
biomedical	ECG, EEG, EMG, SpO2
finance	Market prices, returns, volume
energy	Power grid frequency, load
seismic	Earthquake, vibration sensors
sensors	IoT, industrial sensors
mlops	Model drift, data quality
security	Network traffic, intrusion
industrial	Machinery, SCADA

Guides

• Common Mistakes — Pitfalls and how to avoid them
• Performance Guide — Which endpoint to use, latency tips, async patterns
• Debugging Guide — Troubleshooting errors and unexpected results

Note: These guides are also included in the source distribution (pip install alphainfo and check the package source).

Links

• API Documentation
• Benchmarks
• Dashboard

License

MIT