Cosmic Python SDK 🚀

A comprehensive Python SDK for accessing Cosmic Markets APIs, providing seamless integration with Solana DeFi protocols including Drift Protocol, Jupiter, Kamino, and more.

✨ Features

• 🎯 Drift Protocol V2 - Markets, positions, orders, PnL history, and analytics
• 🔄 Jupiter Integration - DCA orders, limit orders (v1/v2), and token analytics
• 💧 Kamino Finance - Swap data, orderbooks, and analytics
• 📊 OHLC Data - Historical price data for tokens
• 👤 Trader Profiles - Trader scoring and performance metrics
• 🔍 Token Search - Comprehensive token discovery
• ⚡ Async Support - Built on httpx for high-performance asynchronous requests
• 🔒 Type Safe - Fully typed with Pydantic V2 models

📦 Installation

To install directly from GitHub:

pip install git+https://github.com/cosmic-markets/cosmic-python-sdk.git

🚀 Quick Start

Basic Usage

The SDK uses an asynchronous context manager for efficient connection handling.

import asyncio
from cosmic_sdk import Cosmic

async def main():
    # Initialize client (no API key = rate limits apply)
    async with Cosmic() as client:
        # Fetch all Drift markets
        markets = await client.drift.get_all_markets()
        print(f"📊 Found {markets.content[0].name} and {len(markets.content)-1} others")

if __name__ == "__main__":
    asyncio.run(main())

With API Key (Recommended)

To avoid strict rate limits, use an API key from cosmic.markets/api/docs.

import asyncio
from cosmic_sdk import Cosmic

async def main():
    async with Cosmic(api_key="YOUR_API_KEY") as client:
        # Your code here...
        pass

if __name__ == "__main__":
    asyncio.run(main())

📚 Usage Examples

🎯 Drift Protocol

Get Drift Markets

from cosmic_sdk.models.drift import GetDriftMarketsRequest

async def get_drift_markets():
    async with Cosmic() as client:
        # Fetch active perp markets
        request = GetDriftMarketsRequest(pagination=True, page=1, size=10)
        markets = await client.drift.get_all_markets(request)
        
        for market in markets.content:
             print(f"Market: {market.name} (Index: {market.marketIndex})")
             print(f"  Price: ${market.price}")

Get Drift Top Positions

async def get_drift_top_positions():
    async with Cosmic() as client:
        # Get top active positions by notional size
        positions = await client.drift.get_top_active_drift_positions(limit=5)
        
        for pos in positions:
            print(f"Trader: {pos.userPublicKey}")
            print(f"  Market: {pos.marketName}")
            print(f"  Size: {pos.baseAssetAmount} {pos.baseAssetSymbol}")
            print(f"  PnL: ${pos.unrealizedPnl:,.2f}")

Get Drift Orders

from cosmic_sdk.models.drift import GetDriftMarketOrdersRequest

async def get_drift_orders():
    async with Cosmic() as client:
        # Get recent orders for SOL-PERP
        market_pubkey = "8UJgxaiQx5nTrdDgph5FiahMmzduuLTLf5WmsPegYA6W"
        request = GetDriftMarketOrdersRequest(size=5)
        
        orders = await client.drift.get_market_orders(market_pubkey, request)
        
        for order in orders.content:
            print(f"Order #{order.orderId}: {order.direction} {order.baseAssetAmount} @ {order.price}")

Get Drift Trades

from cosmic_sdk.models.drift import GetDriftFilledTradesRequest

async def get_drift_trades():
    async with Cosmic() as client:
        # Get recent filled trades
        request = GetDriftFilledTradesRequest(size=5)
        trades = await client.drift.get_filled_trades(request)
        
        for trade in trades.content:
            print(f"Trade: {trade.action} {trade.baseAssetAmount} {trade.marketName} @ ${trade.price}")

🔄 Jupiter Aggregator

Get Jupiter DCAs

from cosmic_sdk.models.jupiter import GetDcaOrdersRequest

async def get_jupiter_dcas():
    async with Cosmic() as client:
        request = GetDcaOrdersRequest(size=5)
        dcas = await client.jupiter.get_orders(request)
        
        for order in dcas.content:
            print(f"DCA: {order.inputTokenSymbol} -> {order.outputTokenSymbol}")
            print(f"  Every {order.cycleFrequency} seconds")

Get Jupiter Limit Orders V2

from cosmic_sdk.models.jupiter import GetLimitV2OrdersRequest

async def get_jupiter_limit_v2():
    async with Cosmic() as client:
        request = GetLimitV2OrdersRequest(size=5)
        orders = await client.jupiter.get_limit_v2_orders(request)
        
        for order in orders.content:
            print(f"Limit V2: {order.inputTokenSymbol} -> {order.outputTokenSymbol}")
            print(f"  In: {order.inputAmount} -> Out: {order.outputAmount}")

Get Jupiter Limit Orders V1

from cosmic_sdk.models.jupiter import GetLimitOrdersRequest

async def get_jupiter_limit_v1():
    async with Cosmic() as client:
        request = GetLimitOrdersRequest(size=5)
        orders = await client.jupiter.get_limit_orders(request)
        
        for order in orders.content:
            print(f"Limit V1: {order.inputTokenSymbol} -> {order.outputTokenSymbol}")
            print(f"  Maker: {order.maker}")

💧 Kamino Finance

Get Kamino Orderbook

async def get_kamino_orderbook():
    async with Cosmic() as client:
        # Get SOL-USDC Orderbook
        orderbook = await client.kamino.get_sol_usdc_orderbook()
        
        print(f"Top Bid: ${orderbook.bids[0].price} ({orderbook.bids[0].size} SOL)")
        print(f"Top Ask: ${orderbook.asks[0].price} ({orderbook.asks[0].size} SOL)")

Get Top Kamino Trades

from cosmic_sdk.models.kamino import GetKaminoTopFilledTradesRequest

async def get_top_kamino_trades():
    async with Cosmic() as client:
        request = GetKaminoTopFilledTradesRequest(limit=5)
        trades = await client.kamino.get_top_filled_trades(request)
        
        for trade in trades:
            print(f"Swap: {trade.sourceTokenSymbol} -> {trade.destinationTokenSymbol}")
            print(f"  Value: ${trade.filledOutputAmountUsd:,.2f}")

🧩 Modules

Module	Description
drift	Drift Protocol V2 - Markets, positions, orders, history, and analytics
jupiter	Jupiter aggregator - DCA orders, limit orders, and token analytics
kamino	Kamino Finance - Swap data, orderbooks, and analytics
ohlc	Historical OHLC price data for tokens
trader	Trader profiling, scoring, and performance metrics
search	Token search and discovery capabilities

🧪 Testing

To run the test suite, installing pytest and pytest-asyncio is required:

pip install pytest pytest-asyncio

Run unit tests:

pytest tests/

Run integration tests (requires internet connection):

pytest tests/test_integration.py

📄 License

This project is licensed under the MIT License.

🔗 Links

• 🌐 Cosmic Markets
• 📚 API Documentation