ASTRIA — Python SDK

Python SDK for ASTRIA, the Subspace Computing Engine.

Installation

pip install subspacecomputing

Usage

Initialization

from subspacecomputing import ASTRIA

# Initialize the client (defaults to production URL)
client = ASTRIA(api_key='your-api-key-here')

Teams (X-Team-Id)

Keys can carry a default_team_id from the portal. If the user is in several teams, some endpoints require an explicit team: pass team_id=... or call set_team_id so the SDK sends X-Team-Id.

If you omit it when required, the API returns 400 with error.reason team_context_required. Use ValidationError and read exc.reason.

client = ASTRIA(api_key="...", team_id="00000000-0000-0000-0000-000000000000")
client.set_team_id(None)  # clear override for following requests

Simple Projection (1 scenario)

# Create a simple SP Model
spec = {
    'scenarios': 1,  # Must be 1 for /project
    'steps': 12,
    'variables': [
        {
            'name': 'capital',
            'init': 1000.0,
            'formula': 'capital[t-1] * 1.05'  # 5% growth per period
        }
    ]
}

# Run the projection
result = client.project(spec)

# Display results
print(f"Final capital: {result['final_values']['capital']}")
print(f"Trajectory: {result['trajectory']['capital']}")

Monte Carlo Simulation (Multiple scenarios)

# SP Model with random variables
spec = {
    'scenarios': 1000,  # 1000 Monte Carlo scenarios
    'steps': 12,
    'variables': [
        {
            'name': 'taux',
            'dist': 'uniform',
            'params': {'min': 0.03, 'max': 0.07},
            'per': 'scenario'
        },
        {
            'name': 'capital',
            'init': 1000.0,
            'formula': 'capital[t-1] * (1 + taux)'
        }
    ]
}

# Run the simulation
result = client.simulate(spec)

# Analyze results
print(f"Mean final capital: {result['last_mean']['capital']}")
print(f"Median: {result['statistics']['capital']['median']}")
print(f"P5: {result['statistics']['capital']['percentiles']['5']}")
print(f"P95: {result['statistics']['capital']['percentiles']['95']}")

Batch Mode (Multiple Entities)

# SP Model template
template = {
    'scenarios': 1,
    'steps': '65 - batch_params.age',  # Dynamic steps
    'variables': [
        {
            'name': 'age_actuel',
            'init': 'batch_params.age'
        },
        {
            'name': 'salaire',
            'init': 'batch_params.salary',
            'formula': 'salaire[t-1] * 1.03'  # 3% annual increase
        },
        {
            'name': 'capital_retraite',
            'init': 0.0,
            'formula': 'capital_retraite[t-1] * 1.05 + salaire[t] * 0.10'
        }
    ]
}

# Entity data
batch_params = [
    {'entity_id': 'emp_001', 'age': 45, 'salary': 60000},
    {'entity_id': 'emp_002', 'age': 50, 'salary': 80000},
    {'entity_id': 'emp_003', 'age': 35, 'salary': 50000}
]

# Global aggregations (optional)
aggregations = [
    {
        'name': 'capital_total',
        'formula': 'sum(capital_retraite[t_final])'
    },
    {
        'name': 'moyenne_capital',
        'formula': 'mean(capital_retraite[t_final])'
    }
]

# Run batch
result = client.project_batch(
    template=template,
    batch_params=batch_params,
    aggregations=aggregations
)

# Analyze results
for entity in result['entities']:
    print(f"{entity['_entity_id']}: Capital = {entity['final_values']['capital_retraite']}")

print(f"Total capital: {result['aggregations']['capital_total']}")
print(f"Average: {result['aggregations']['moyenne_capital']}")

Run a stored model (projection_id + inputs)

Execute a projection already saved in Subspace. Load run_inputs once (cache in your app), build UI/batch from that descriptor, then call run_model many times with only the overrides. Astria runs the calculation; model structure (formula, dist type, per) is never overridable.

proj = client.load_model(projection_id)  # get_projection + in-memory cache on this client
# Or: proj = client.get_projection(projection_id)

result = client.run_model(projection_id, inputs={"capital": 25000})
print(result["final_values"]["capital"])

Input shapes by run_inputs.kind:

kind	Example value in inputs
init	{"capital": 25000}
params	{"taux": {"min": 0.02, "max": 0.05}}
values	{"inflation": [0.01, 0.02, 0.03]}
table_ref / table	{"mortality": "<uuid>"} or {"grid": {...}}
run	{"scenarios": 100, "steps": 12}

Unknown input keys → API 400 unknown_run_input. rerun_run(run_id) re-executes a past run snapshot, not the live model.

Pattern for a dynamic form / batch loop:

proj = client.load_model(projection_id)
inputs = {}
for desc in proj.get("run_inputs") or []:
    kind = desc.get("kind")
    name = desc["name"]
    if kind == "init":
        inputs[name] = desc.get("default")  # or from UI
    elif kind == "params":
        inputs[name] = dict(desc.get("default") or {})
    # elif kind == "values": ...
result = client.run_model(projection_id, inputs=inputs)

Runs: read, artifact, rerun, replay

Four distinct operations on persisted runs:

Method	What it does
get_run(run_id)	Read metadata, stored SP Model snapshot, and result_summary
get_run_artifact(artifact_id)	Read heavy results already computed (final_values / URL)
rerun_run(run_id, spec_patch=None)	Re-execute the run's snapshot in Astria; creates a new run
replay_run(run_id, scenario_id)	Reproduce one Monte Carlo scenario when seeds were stored

# Cross-language analysis: app computed → Python reads original results
run = client.get_run(run_id)
print(run["result_summary"])
if run.get("artefact_id"):
    artifact = client.get_run_artifact(run["artefact_id"])
    print(artifact.get("final_values"))

# Re-execute the exact snapshot (validation) — new run, same projection
baseline = client.rerun_run(run_id)
print(baseline["run_id"], baseline["source_spec_hash"], baseline["spec_changed"])

# What-if: fork the snapshot with a JSON Merge Patch (source run unchanged)
what_if = client.rerun_run(
    run_id,
    spec_patch={"meta": {"seed": 42}},
)
print(what_if["executed_spec_hash"], what_if["final_values"])

# Exact MC scenario reproduction (requires meta.store_seeds_for_replay at creation)
replay = client.replay_run(run_id, scenario_id=0)

Notes

• rerun_run does not load the projection's current dsl_template; it uses the run's stored spec.
• Requires persist_mode=full (headless minimal/metadata keys cannot create a new run).
• Exact stochastic reproduction of one scenario uses replay_run + stored seeds; a plain rerun_run of a Monte Carlo may draw new samples unless meta.seed was in the snapshot.
• Live table_refs may resolve to newer data than at original execution time.

Validation

# Validate an SP Model before execution
validation = client.validate(spec)

if validation['is_valid']:
    print("✅ SP Model is valid")
    if validation.get('warnings'):
        print(f"⚠️  Warnings: {validation['warnings']}")
else:
    print(f"❌ Errors: {validation['errors']}")

Utilities

# Get examples
examples = client.get_examples()
print(f"Available examples: {len(examples['examples'])}")

# Check usage
usage = client.get_usage()
print(f"Simulations used: {usage['usage']['simulations_used']}/{usage['usage']['simulations_limit']}")

# Get plans
plans = client.get_plans()
for plan in plans['plans']:
    print(f"{plan['name']}: ${plan['price_monthly']}/month")

Error Handling

from subspacecomputing import (
    Subspace,
    SubspaceError,
    QuotaExceededError,
    RateLimitError,
    AuthenticationError,
    ValidationError,
)

try:
    result = client.simulate(spec)
except QuotaExceededError as e:
    print(f"Monthly quota exceeded: {e}")
except RateLimitError as e:
    print(f"Rate limit exceeded: {e}")
    print(f"Retry after: {e.response.headers.get('Retry-After')} seconds")
except AuthenticationError as e:
    print(f"Invalid API key: {e}")
except ValidationError as e:
    print(f"Validation error: {e.detail}")
except SubspaceError as e:
    print(f"API error: {e}")

Rate Limit and Quota Information

After making a request, you can check your rate limit and quota status:

# Make a request
result = client.project(spec)

# Check rate limit info
rate_limit = client.get_rate_limit_info()
if rate_limit:
    print(f"Rate limit: {rate_limit['remaining']}/{rate_limit['limit']} remaining")

# Check quota info
quota = client.get_quota_info()
if quota:
    print(f"Quota: {quota['used']}/{quota['limit']} used, {quota['remaining']} remaining")

Documentation

Check out the full documentation at https://www.subspacecomputing.com/developer

API reference is available at https://www.subspacecomputing.com/docs

For support, reach out to contact@subspacecomputing.com

License

MIT License. Check the LICENSE file for details.

Copyright

© 2026 Subspace Computing Inc. All Rights Reserved