MangroveAI Python SDK

Python SDK for the MangroveAI trading strategy platform.

Install

pip install mangroveai

The package imports as mangrove_ai (e.g. from mangrove_ai import MangroveAI). The legacy import mangroveai path also still works for backwards compatibility.

Setup

1. Create an account at mangrovedeveloper.ai
2. Navigate to Settings > API Keys
3. Generate a new API key
4. Set it as an environment variable:

macOS / Linux (bash/zsh):

export MANGROVE_API_KEY=prod_your_key_here

Windows (PowerShell):

$env:MANGROVE_API_KEY = "prod_your_key_here"

Windows (cmd.exe):

set MANGROVE_API_KEY=prod_your_key_here

Or use a .env file

The SDK can load a .env file automatically if you install the optional python-dotenv extra:

pip install "mangroveai[dotenv]"

Then create a .env file in your project root:

MANGROVE_API_KEY=prod_your_key_here

MangroveAI() will pick it up with no further code. Real environment variables always take precedence over .env values, so a .env file never overrides a key you've already exported. Disable autoloading with MangroveAI(load_dotenv=False). If python-dotenv isn't installed, the SDK reads os.environ only (a .env file is ignored).

Quickstart

from mangrove_ai import MangroveAI

client = MangroveAI()  # reads MANGROVE_API_KEY from environment

# List trading signals
signals = client.signals.list(limit=10)
for s in signals.items:
    print(f"{s.name} ({s.category}, {s.signal_type})")

# Get market data
btc = client.crypto_assets.get_market_data("BTC")
print(f"BTC: ${btc.data['current_price']:,.2f}")

# Create a strategy
from mangrove_ai.models import CreateStrategyRequest

strategy = client.strategies.create(CreateStrategyRequest(
    name="RSI Momentum",
    asset="BTC",
    entry=[{"name": "rsi_oversold", "signal_type": "TRIGGER",
            "timeframe": "1d", "params": {"window": 14, "threshold": 30}}],
))

# Run a backtest (blocks until done; async-backed since v1.14, so long
# lookbacks work -- it submits to the async surface and polls internally)
import json
from mangrove_ai.models import BacktestRequest

result = client.backtesting.run(BacktestRequest(
    asset="BTC",
    interval="1d",
    strategy_json=json.dumps({"name": "test", "asset": "BTC",
        "entry": [{"name": "rsi_oversold", "signal_type": "TRIGGER",
                    "timeframe": "1d", "params": {"window": 14, "threshold": 30}}],
        "exit": []}),
    lookback_months=3,
    initial_balance=10000,
    min_balance_threshold=0.1, min_trade_amount=25,
    max_open_positions=3, max_trades_per_day=10,
    max_risk_per_trade=0.02, max_units_per_trade=1000000,
    max_trade_amount=10000000, volatility_window=24,
    target_volatility=0.1,
    # Optional: per-timeframe cooldown configuration (preferred over legacy flat fields).
    # Keys are the primary timeframe; each value carries max_hold_time_hours,
    # short_loss_limit, long_loss_limit, short_window_bars, and long_window_bars.
    cooldown_config={
        "1d": {
            "max_hold_time_hours": 10,
            "short_loss_limit": 4,
            "long_loss_limit": 6,
            "short_window_bars": 20,
            "long_window_bars": 60,
        }
    },
))
print(f"Trades: {result.trade_count}, Sharpe: {result.metrics.get('sharpe_ratio')}")

Services

Layer 1: MangroveAI Core API

Layer 2: Knowledge Base API

On-chain capability surface

client.on_chain covers Mangrove's Nansen Pro plan plus WhaleAlert (Developer tier, 30-day history):

On-chain time series → signals. get_onchain_series returns a per-bar series for any window — the same call serves a live trailing window (e.g. last 10 days, ending now) or a long historical range. Build a DataFrame with pd.DataFrame(resp.series).set_index("timestamp") and feed it to a mangrove-kb on-chain signal. End-to-end walkthrough: examples/onchain_signals_demo.py.

filters and order_by pass through directly to the upstream Nansen API — restrict by include_smart_money_labels, set value_usd min/max bounds, sort by any field. See examples/on_chain_nansen.py for raw-method snippets.

Oracle — SIEVE + sweep

client.oracle.* is Mangrove's strategy-research engine. Two headline tools:

• SIEVE scores up to 99 strategy ideas in one call and tells you which are worth testing — a binary go/no-go (p_trades vs p_no_trades) plus a 4-class outcome head (losing / no_trades / wash / winning). Most ideas never trade meaningfully; SIEVE finds the ones that do before you spend compute backtesting them.
• Sweep fans a parameter search into one managed experiment of many backtests and ranks the results.

Score ideas with SIEVE

from mangrove_ai import MangroveAI
from mangrove_ai.models.oracle import SieveScoreRequest, StrategyInput, SignalSpec

client = MangroveAI()  # reads MANGROVE_API_KEY

resp = client.oracle.sieve_score(SieveScoreRequest(strategies=[
    StrategyInput(
        asset="BTC",
        entry=[SignalSpec(name="ema_crossover", signal_type="TRIGGER", timeframe="1h")],
        exit=[SignalSpec(name="rsi", signal_type="FILTER", timeframe="1h")],
    ),
    # ...up to 99 strategies per request
]))

print(f"scored {resp.count} | model {resp.model_version}")
for p in resp.predictions:
    print(f"  go/no-go: {p.binary}")      # {'p_no_trades': .., 'p_trades': ..}
    print(f"  outcome:  {p.four_class}")  # {'losing':.., 'no_trades':.., 'wash':.., 'winning':..}

Run a parameter sweep (experiment lifecycle)

A sweep is a draft experiment you validate, then launch; it fans out into backtests asynchronously. Poll get_experiment / list_results to track it.

# Pick a dataset + execution defaults from the API, then build a config
ds = client.oracle.list_datasets()[0]
exec_config = client.oracle.exec_config_defaults()

config = {
    "name": "ema-sweep-demo",
    "kind": "single",
    "search_mode": "random",
    "seed": 42,
    "n_random": 50,                 # number of random strategies to draw
    "datasets": [ds],
    "random_signals": {
        "n_entry_triggers": 1, "min_entry_filters": 0, "max_entry_filters": 2,
        "min_exit_triggers": 0, "max_exit_triggers": 1,
        "min_exit_filters": 0, "max_exit_filters": 1,
        "n_param_draws": 3, "allowed_categories": None,
    },
    "execution_config": exec_config,
}

created = client.oracle.create_experiment(config)            # -> status "draft"
val = client.oracle.validate_experiment(created.experiment_id)
print(f"valid={val.valid} total_runs={val.total_runs} errors={val.errors}")

if val.valid:                                                # must pass before launch
    client.oracle.launch_experiment(created.experiment_id)   # fans out asynchronously
    status = client.oracle.get_experiment(created.experiment_id)
    print(f"status={status['status']} completed={status.get('completed_runs')}")
    # client.oracle.pause_experiment(id) / delete_experiment(id) to stop or clean up

results = client.oracle.list_results(experiment_id=created.experiment_id, limit=20)

Other client.oracle.* methods

• Backtests: backtest, backtest_async + backtest_poll, backtest_bulk
• Corpus query: data_query (curated BigQuery proxy over the results corpus)
• Simulation: simulate_run, simulate_generate, simulate_presets, simulate_history
• Results & catalogs: list_results, list_datasets, list_signals, list_templates, exec_config_defaults
• Leaderboard & live: leaderboard, list_deployed_strategies, get_deployed_strategy_state, get_deployed_strategy_events

Full reference: SIEVE pre-filter guide, SIEVE end-to-end, and the Experiments API reference.

Environment Detection

The SDK auto-detects the environment from your API key prefix:

Override with explicit parameters:

client = MangroveAI(api_key="...", base_url="http://localhost:5001/api/v1")

Error Handling

from mangrove_ai import MangroveAI, NotFoundError, RateLimitError, APIError

client = MangroveAI()

try:
    strategy = client.strategies.get("nonexistent-id")
except NotFoundError as e:
    print(f"Not found: {e.message} (correlation_id={e.correlation_id})")
except RateLimitError as e:
    print(f"Rate limited, retry after {e.retry_after}s")
except APIError as e:
    print(f"[{e.status_code}] {e.code}: {e.message}")

Pagination

Paginated endpoints return PaginatedResponse[T]:

# Single page
page = client.strategies.list(skip=0, limit=10)
print(f"Showing {len(page.items)} of {page.total}")

# Auto-paginate all items
for strategy in client.strategies.list_iter():
    print(strategy.name)

Examples

See the examples/ directory for working scripts.

Development

git clone https://github.com/MangroveTechnologies/mangrove-ai-sdk.git
cd mangrove-ai-sdk
pip install -e ".[dev]"
pytest tests/ --ignore=tests/integration  # unit tests
MANGROVE_API_KEY=... pytest tests/integration/ -m integration  # live tests