Code Examples

Working Python samples for the NDinomics Bazaar API.

Prerequisites: pip install requests
Start a local Bazaar instance: python -m uvicorn api.main:app --port 8000
All examples default to BASE_URL = "http://localhost:8000".

Example	Complexity	What It Demonstrates
Simple Strategy	Simple	Full strategy lifecycle: register, submit, gauntlet, publish
Signal Feed	Medium	Register signal, submit 30 days of data, check accuracy
Complete Skill	Complete	Build, execute, review, and publish an analysis skill

Simple Strategy Simple

Full lifecycle of a 3-ticker conservative strategy: register an account, submit the allocation, run the gauntlet stress test, check the nutrition label, and publish to the marketplace.

"""
Simple Strategy Example — NDinomics Bazaar Value Adder API
Full lifecycle: register, submit, gauntlet, nutrition label, publish
"""
import requests

BASE_URL = "http://localhost:8000"

def main():
    # Step 1: Register a value adder account
    print("1. Registering value adder account...")
    resp = requests.post(f"{BASE_URL}/accounts/v1/register", json={
        "email": "advisor@example.com",
        "name": "Example Advisory LLC",
        "account_type": "value_adder",
    })
    resp.raise_for_status()
    account = resp.json()
    api_key = account["api_key"]
    account_id = account["account_id"]
    headers = {"Authorization": f"Bearer {api_key}"}

    # Step 2: Submit a strategy
    print("2. Submitting conservative growth strategy...")
    resp = requests.post(f"{BASE_URL}/strategies/v1/submit", headers=headers, json={
        "strategy_id": "conservative-growth-v1",
        "name": "Conservative Growth",
        "publisher": account_id,
        "description": "60/30/10 core allocation",
        "allocation": {
            "VTI": 0.60,  # 60% US total stock market
            "BND": 0.30,  # 30% US aggregate bond
            "GLD": 0.10,  # 10% gold (inflation hedge)
        },
        "rebalance": {
            "trigger": "drift_pct",
            "threshold": 10.0,
            "cadence": "quarterly",
        },
    })
    resp.raise_for_status()

    # Step 3: Run the gauntlet (stress test)
    print("3. Running gauntlet stress test...")
    resp = requests.post(
        f"{BASE_URL}/strategies/v1/conservative-growth-v1/gauntlet",
        headers=headers,
    )
    resp.raise_for_status()
    gauntlet = resp.json()
    print(f"   Passed: {gauntlet['gauntlet_passed']}")

    # Step 4: Check nutrition label
    print("4. Fetching nutrition label...")
    resp = requests.get(
        f"{BASE_URL}/strategies/v1/conservative-growth-v1/nutrition-label",
        headers=headers,
    )

    # Step 5: Publish (only if gauntlet passed)
    if gauntlet["gauntlet_passed"]:
        print("5. Publishing strategy to marketplace...")
        resp = requests.post(
            f"{BASE_URL}/strategies/v1/conservative-growth-v1/publish",
            headers=headers,
        )
        resp.raise_for_status()
        print("   Published!")

if __name__ == "__main__":
    main()

Signal Feed Medium

Register a sentiment signal feed covering 5 tech mega-caps, submit 30 days of daily data points in batch, then check accuracy and promotion eligibility.

"""
Signal Feed Example — register, submit daily data, check accuracy
"""
import random
from datetime import datetime, timedelta
import requests

BASE_URL = "http://localhost:8000"
TICKERS = ["AAPL", "MSFT", "GOOGL", "AMZN", "NVDA"]

def main():
    # Register account
    resp = requests.post(f"{BASE_URL}/accounts/v1/register", json={
        "email": "quant@example.com",
        "name": "Quant Signals Inc",
        "account_type": "value_adder",
    })
    resp.raise_for_status()
    account = resp.json()
    headers = {"Authorization": f"Bearer {account['api_key']}"}

    # Register sentiment signal
    resp = requests.post(f"{BASE_URL}/signals/v1/register", headers=headers, json={
        "signal_id": "social-sentiment-tech-mega",
        "signal_type": "sentiment",
        "description": "Social media sentiment for tech mega-caps",
        "asset_coverage": TICKERS,
        "update_frequency": "daily",
        "schema": {
            "score_range": [-1.0, 1.0],
            "confidence_range": [0.0, 1.0],
            "direction_values": ["up", "down", "neutral"],
        },
    })
    resp.raise_for_status()

    # Submit 30 days of daily data
    base_date = datetime(2026, 4, 1, 14, 30, 0)
    for day in range(30):
        timestamp = (base_date + timedelta(days=day)).isoformat() + "Z"
        points = []
        for ticker in TICKERS:
            score = round(random.uniform(-0.8, 0.9), 3)
            confidence = round(random.uniform(0.5, 0.95), 2)
            direction = "up" if score > 0.2 else ("down" if score < -0.2 else "neutral")
            points.append({
                "ticker": ticker, "score": score,
                "confidence": confidence, "timestamp": timestamp,
                "direction": direction,
            })
        requests.post(
            f"{BASE_URL}/signals/v1/social-sentiment-tech-mega/submit",
            headers=headers, json={"points": points},
        ).raise_for_status()

    # Check accuracy
    resp = requests.get(
        f"{BASE_URL}/signals/v1/social-sentiment-tech-mega/accuracy",
        headers=headers,
    )
    accuracy = resp.json()
    print(f"Accuracy: {accuracy.get('accuracy', {})}")
    print(f"Promotion eligible: {accuracy.get('promotion_eligibility', {})}")

if __name__ == "__main__":
    main()

Complete Skill Complete

Full skill lifecycle: define analysis logic, register on the platform, execute against sample portfolio data, review recommendations, and publish to the marketplace.

"""
Complete Skill Example — build, execute, and publish a sector rotation analyzer
"""
import requests

BASE_URL = "http://localhost:8000"

# Analysis function (runs in platform sandbox)
SKILL_CODE = """
def analyze(input_data):
    holdings = input_data.get("holdings", {})
    sector_grades = input_data.get("sector_grades", {})
    regime = input_data.get("regime", "unknown")

    sector_etfs = {
        "technology": "XLK", "healthcare": "XLV",
        "energy": "XLE", "financials": "XLF",
    }
    defensive = {"healthcare", "utilities"}
    grade_score = {"A": 4, "B+": 3.5, "B": 3, "C+": 2.5, "C": 2, "D": 1, "F": 0}

    recommendations = []
    for sector, grade in sector_grades.items():
        etf = sector_etfs.get(sector)
        if not etf: continue
        score = grade_score.get(grade, 2)
        if regime == "contraction" and sector in defensive:
            score += 1.0
        current_weight = holdings.get(etf, 0.0)
        if score >= 3.5 and current_weight < 0.15:
            recommendations.append({
                "action": "overweight", "ticker": etf,
                "rationale": f"{sector} grade {grade} in {regime}",
                "estimated_impact": {"return_delta": 0.01, "risk_delta": 0.005},
            })
    return recommendations
"""

def main():
    # Register account
    resp = requests.post(f"{BASE_URL}/accounts/v1/register", json={
        "email": "quant-dev@example.com",
        "name": "Rotation Capital",
        "account_type": "value_adder",
    })
    resp.raise_for_status()
    account = resp.json()
    headers = {"Authorization": f"Bearer {account['api_key']}"}

    # Register skill
    resp = requests.post(f"{BASE_URL}/skills/v1/register", headers=headers, json={
        "skill_id": "sector-rotation-v1",
        "name": "Sector Rotation Analyzer",
        "publisher": account["account_id"],
        "skill_type": "analyzer",
        "input_requires": ["holdings", "sector_grades", "regime"],
        "output_schema": {
            "type": "recommendations",
            "fields": {
                "action": "overweight | underweight | hold",
                "ticker": "string",
                "rationale": "string",
            },
        },
        "skill_code": SKILL_CODE,
    })
    resp.raise_for_status()

    # Execute against sample portfolio
    resp = requests.post(
        f"{BASE_URL}/skills/v1/sector-rotation-v1/execute",
        headers=headers, json={
            "account_id": "demo_user_001",
            "consent_ref": "consent_demo",
            "input_data": {
                "holdings": {"VTI": 0.40, "QQQ": 0.25, "BND": 0.20},
                "sector_grades": {"technology": "B+", "healthcare": "A"},
                "regime": "expansion",
            },
        },
    )
    execution = resp.json()
    for rec in execution.get("recommendations", []):
        print(f"  {rec['action'].upper()} {rec['ticker']}: {rec['rationale']}")

    # Publish
    resp = requests.post(
        f"{BASE_URL}/skills/v1/sector-rotation-v1/publish",
        headers=headers,
    )
    if resp.status_code == 200:
        print("Published to marketplace!")

if __name__ == "__main__":
    main()

Key Rules

Strategies: weights must sum to 1.0, no leverage, no negative weights
Signals: start in shadow mode, minimum 1 quarter before promotion
Skills: require explicit user consent for data access, output always labeled third-party
All items must pass the gauntlet before publication
API keys are shown once at registration — store them securely

Full API Reference → Getting Started → Open Studio →