Skip to content

Python Integration Examples

This section provides examples of how to integrate with the API using Python.

Authentication

import hashlib
import hmac
import time
import requests
import json

class ApiClient:
    def __init__(self, base_url, access_key, secret_key):
        self.base_url = base_url.rstrip('/')
        self.access_key = access_key
        self.secret_key = secret_key

    def _get_timestamp(self):
        return str(int(time.time() * 1000))

    def _generate_signature(self, path, timestamp, body=''):
        message = f"{self.access_key}{path}{timestamp}{body}"
        signature = hmac.new(
            self.secret_key.encode('utf-8'),
            message.encode('utf-8'),
            hashlib.sha256
        ).hexdigest()
        return signature

    def request(self, method, endpoint, params=None, data=None):
        path = f"/api/v1{endpoint}"
        url = f"{self.base_url}{path}"

        timestamp = self._get_timestamp()

        headers = {
            'X-Access-Key': self.access_key,
            'X-Timestamp': timestamp,
        }

        if method.upper() in ['POST', 'PUT', 'PATCH'] and data is not None:
            body = json.dumps(data)
            headers['Content-Type'] = 'application/json'
        else:
            body = ''

        headers['X-Signature'] = self._generate_signature(path, timestamp, body)

        response = requests.request(
            method=method,
            url=url,
            params=params,
            data=body if body else None,
            headers=headers
        )

        return response.json()

    def get(self, endpoint, params=None):
        return self.request('GET', endpoint, params=params)

    def post(self, endpoint, data=None):
        return self.request('POST', endpoint, data=data)

Usage Examples

Check Server Status

api = ApiClient('https://api.001k.world', 'your_access_key', 'your_secret_key')

# Status endpoint doesn't require authentication
status = requests.get('https://api.001k.world/api/v1/status').json()
print(f"Server timestamp: {status['timestamp']}")

Get Account Balances

api = ApiClient('https://api.001k.world', 'your_access_key', 'your_secret_key')

balances = api.get('/balance')
for balance in balances:
    print(f"{balance['currency']}: {balance['available']} (available), {balance['frozen']} (frozen)")

Generate Deposit Address

api = ApiClient('https://api.001k.world', 'your_access_key', 'your_secret_key')

response = api.post('/deposit/generate_address', {
    'currency': 'BTC',
    'method': 'btc_native',
    'callback_url': 'https://your-server.com/callbacks/deposits',
    'custom_id': 'my-deposit-address-123'
})

print(f"Deposit address: {response['address']}")
if response.get('tag'):
    print(f"Tag: {response['tag']}")

Create Withdrawal

api = ApiClient('https://api.001k.world', 'your_access_key', 'your_secret_key')

response = api.post('/withdraw', {
    'currency': 'BTC',
    'method': 'btc_native',
    'amount': '0.01000000',
    'address': 'bc1qxy2kgdygjrsqtzq2n0yrf2493p83kkfjhx0wlh',
    'callback_url': 'https://your-server.com/callbacks/withdrawals',
    'custom_id': 'my-withdrawal-123'
})

print(f"Withdrawal created: {response['code']}")
print(f"Status: {response['status']}")

Create Swap

api = ApiClient('https://api.001k.world', 'your_access_key', 'your_secret_key')

# First, get available swap directions
directions = api.get('/swap-directions')
for direction in directions:
    print(f"{direction['code']}: {direction['spend_currency']} -> {direction['receive_currency']} (rate: {direction['rate']})")

# Create a swap
response = api.post('/swap/open', {
    'direction': 'BTC-USDT',
    'spend_amount': '0.01000000',
    'callback_url': 'https://your-server.com/callbacks/swaps',
    'custom_id': 'my-swap-123'
})

print(f"Swap created: {response['code']}")
print(f"Spend: {response['spend_amount']} {response['spend_currency']}")
print(f"Receive: {response['receive_amount']} {response['receive_currency']}")
print(f"Status: {response['status']}")

Create Internal Transfer

api = ApiClient('https://api.001k.world', 'your_access_key', 'your_secret_key')

response = api.post('/internal-transfer', {
    'currency_code': 'BTC',
    'amount': '0.01000000',
    'recipient': 'user123',
    'callback_url': 'https://your-server.com/callbacks/transfers',
    'custom_id': 'my-transfer-123',
    'note': 'Payment for services'
})

print(f"Transfer created: {response['code']}")
print(f"Status: {response['status']}")

Create Limit Order

api = ApiClient('https://api.001k.world', 'your_access_key', 'your_secret_key')

# First, get available trading pairs
pairs = api.get('/limit-pairs')
for pair in pairs:
    print(f"{pair['symbol']}: {pair['current_price']} (fee: {pair['fee_percent']}%)")

# Create a limit order
response = api.post('/limit-order', {
    'pair': 'BTCUSDT',
    'side': 'buy',
    'receive_amount': '0.01000000',
    'limit_price': '29000.00',
    'callback_url': 'https://your-server.com/callbacks/orders',
    'custom_id': 'my-order-123'
})

print(f"Order created: {response['code']}")
print(f"Status: {response['status']}")
print(f"Base amount: {response['base_amount']}")
print(f"Quote amount: {response['quote_amount']}")
print(f"Filled base amount: {response['filled_base_amount']}")
print(f"Filled quote amount: {response['filled_quote_amount']}")
print(f"Fee currency: {response['fee_currency']}")
print(f"Filled fee amount: {response['filled_fee_amount']}")
print(f"Created at: {response['created_at']}")
print(f"Updated at: {response['updated_at']}")
print(f"Done at: {response.get('done_at', 'N/A')}")

Create AML Check

api = ApiClient('https://api.001k.world', 'your_access_key', 'your_secret_key')

# First, get available plans and tokens
plans = api.get('/aml/plan')
tokens = api.get('/aml/token')

for plan in plans:
    print(f"Plan: {plan['code']} - {plan['name']} (price: {plan['price']} {plan['currency']})")

for token in tokens:
    print(f"Token: {token['symbol']} on {token['network']} (ID: {token['id']})")

# Create an AML check
response = api.post('/aml/check', {
    'direction': 'withdrawal',
    'token': 1,  # Bitcoin
    'address': 'bc1qxy2kgdygjrsqtzq2n0yrf2493p83kkfjhx0wlh',
    'callback_url': 'https://your-server.com/callbacks/aml',
    'custom_id': 'my-aml-check-123'
})

print(f"AML check created: {response['code']}")
print(f"Status: {response['status']}")
print(f"Direction: {response['direction']}")
print(f"Token: {response['token']}")
print(f"Address: {response['address']}")
print(f"Transaction: {response.get('transaction', 'N/A')}")
print(f"Created at: {response['created_at']}")
print(f"Updated at: {response['updated_at']}")
if response.get('report'):
    print(f"Report status: {response['report']['status']}")
    print(f"Risk score: {response['report']['riskscore']}")
    print(f"Alert grade: {response['report']['alert_grade']}")

Callback Handling

When the status of an operation changes, the API will send a callback to the URL provided in the callback_url parameter. Here's an example of how to handle these callbacks in a Flask application:

Important Note: You may not receive callbacks for every intermediate status change or field update if changes occur very rapidly. When operations transition through multiple states quickly, intermediate callbacks may be skipped. However, you will always receive callbacks for the final states of operations. Design your callback handling to be idempotent and not dependent on receiving every single state transition.

from flask import Flask, request, jsonify
import requests
from functools import wraps

app = Flask(__name__)

# Your API client
api = ApiClient('https://api.001k.world', 'your_access_key', 'your_secret_key')

# IP whitelist for callback security
ALLOWED_IPS = ['192.168.1.1', '10.0.0.1']  # Replace with actual IPs provided by support

def verify_ip(f):
    @wraps(f)
    def decorated_function(*args, **kwargs):
        client_ip = request.remote_addr
        if client_ip not in ALLOWED_IPS:
            return jsonify({'error': 'Unauthorized IP'}), 403
        return f(*args, **kwargs)
    return decorated_function

def verify_data(endpoint, code):
    """Verify callback data by making an API request to get the current status"""
    try:
        # Make an authenticated API request to verify the status
        response = api.get(f'{endpoint}/{code}')
        return response
    except Exception as e:
        # Handle any API request errors
        print(f"Error verifying data: {e}")
        return None

@app.route('/callbacks/deposits', methods=['POST'])
@verify_ip
def deposit_callback():
    data = request.json
    code = data.get('code')
    status = data.get('status')
    custom_id = data.get('custom_id')

    print(f"Deposit callback received: {code} ({custom_id}) - Status: {status}")

    # Verify the data by making an API request
    verified_data = verify_data('/deposit', code)
    if verified_data and verified_data.get('status') == status:
        # Process the callback (update database, send notifications, etc.)
        print("Deposit status verified")
    else:
        print("Warning: Callback data verification failed!")

    return jsonify({'success': True})

@app.route('/callbacks/withdrawals', methods=['POST'])
@verify_ip
def withdrawal_callback():
    data = request.json
    code = data.get('code')
    status = data.get('status')
    custom_id = data.get('custom_id')

    print(f"Withdrawal callback received: {code} ({custom_id}) - Status: {status}")

    # Verify the data by making an API request
    verified_data = verify_data('/withdraw', code)
    if verified_data and verified_data.get('status') == status:
        # Process the callback
        print("Withdrawal status verified")
    else:
        print("Warning: Callback data verification failed!")

    return jsonify({'success': True})

@app.route('/callbacks/swaps', methods=['POST'])
@verify_ip
def swap_callback():
    data = request.json
    code = data.get('code')
    status = data.get('status')
    custom_id = data.get('custom_id')

    print(f"Swap callback received: {code} ({custom_id}) - Status: {status}")

    # Verify the data by making an API request
    verified_data = verify_data('/swap', code)
    if verified_data and verified_data.get('status') == status:
        # Process the callback
        print("Swap status verified")
    else:
        print("Warning: Callback data verification failed!")

    return jsonify({'success': True})

@app.route('/callbacks/transfers', methods=['POST'])
@verify_ip
def transfer_callback():
    data = request.json
    code = data.get('code')
    status = data.get('status')
    custom_id = data.get('custom_id')

    print(f"Transfer callback received: {code} ({custom_id}) - Status: {status}")

    # Verify the data by making an API request
    verified_data = verify_data('/internal-transfer', code)
    if verified_data and verified_data.get('status') == status:
        # Process the callback
        print("Transfer status verified")
    else:
        print("Warning: Callback data verification failed!")

    return jsonify({'success': True})

@app.route('/callbacks/orders', methods=['POST'])
@verify_ip
def order_callback():
    data = request.json
    code = data.get('code')
    status = data.get('status')
    custom_id = data.get('custom_id')

    print(f"Order callback received: {code} ({custom_id}) - Status: {status}")

    # Verify the data by making an API request
    verified_data = verify_data('/limit-order', code)
    if verified_data and verified_data.get('status') == status:
        # Process the callback
        print("Order status verified")
    else:
        print("Warning: Callback data verification failed!")

    return jsonify({'success': True})

@app.route('/callbacks/aml', methods=['POST'])
@verify_ip
def aml_callback():
    data = request.json
    code = data.get('code')
    status = data.get('status')
    custom_id = data.get('custom_id')

    print(f"AML check callback received: {code} ({custom_id}) - Status: {status}")

    # Verify the data by making an API request
    verified_data = verify_data('/aml/check', code)
    if verified_data and verified_data.get('status') == status:
        # Process the callback
        print("AML check status verified")
    else:
        print("Warning: Callback data verification failed!")

    return jsonify({'success': True})

if __name__ == '__main__':
    app.run(debug=True, port=5000)