Live Management

from base64 import b64encode
from hashlib import sha256
from time import time
from requests import get, post
BASE_URL = 'https://www.quantconnect.com/api/v2/'

# You need to replace these with your actual credentials.
# You can request your credentials at https://www.quantconnect.com/settings/
# You can find our organization ID at https://www.quantconnect.com/organization/ 
USER_ID = 0
API_TOKEN = '____'
ORGANIZATION_ID = '____'

def get_headers():
    # Get timestamp
    timestamp = f'{int(time())}'
    time_stamped_token = f'{API_TOKEN}:{timestamp}'.encode('utf-8')

    # Get hased API token
    hashed_token = sha256(time_stamped_token).hexdigest()
    authentication = f'{USER_ID}:{hashed_token}'.encode('utf-8')
    authentication = b64encode(authentication).decode('ascii')

    # Create headers dictionary.
    return {
        'Authorization': f'Basic {authentication}',
        'Timestamp': timestamp
    }

# Authenticate to verify credentials
response = post(f'{BASE_URL}/authenticate', headers = get_headers())
print(response.json())

# --------------------


### Create Live Algorithm
# Define placeholder IDs for compilation and node (replace with actual values)
project_id = 12345678
compile_id = "compile_id..."
node_id = "node_id..."
# Prepare the data payload for creating a live algorithm with necessary details
payload = {
    "versionId": "-1",  # Use the latest version of the algorithm
    "projectId": project_id,  # ID of the project to deploy as a live algorithm
    "compileId": compile_id,  # Compilation ID for the algorithm code
    "nodeId": node_id,  # Node ID where the algorithm will run
    "brokerage": {  # Brokerage configuration for live trading
        "id": "QuantConnectBrokerage",  # Brokerage identifier
        "user": "",  # Brokerage username (replace with actual value)
        "password": "",  # Brokerage password (replace with actual value)
        "environment": "live-paper",  # Trading environment (live or paper)
        "account": ""  # Brokerage account ID (replace with actual value)
    },
    "dataProviders": {  # Data provider configuration
        "QuantConnectBrokerage": {
            "id": "QuantConnectBrokerage"  # Data provider identifier
        }
    },
    "parameters": {},  # Optional algorithm parameters (empty in this example)
    "notification": {}  # Optional notification settings (empty in this example)
}
# Send a POST request to the /live/create endpoint to deploy the algorithm
response = post(f'{BASE_URL}/live/create', headers=get_headers(), json=data)
# Parse the JSON response into python managable dict from the API
result = response.json()
# Extract the deploy ID from the response for future operations
deploy_id = result['deployId']
# Check if the request was successful and print the result
if result['success']:
    print("Live Algorithm Created Successfully:")
    print(result)

### Read Live Algorithm Statistics
# Prepare data payload with project and deploy IDs to fetch statistics
payload = {
    "projectId": project_id,  # ID of the project
    "deployId": deploy_id  # ID of the deployed live algorithm
}
# Send a POST request to the /live/read endpoint to get algorithm statistics
response = post(f'{BASE_URL}/live/read', headers=get_headers(), json=payload)
# Parse the JSON response into python managable dict
result = response.json()
# Check if the request was successful and print the statistics
if result['success']:
    print("Live Algorithm Statistics:")
    print(result)

### Liquidate Live Algorithm
# Prepare data payload with project ID to liquidate the algorithm
payload = {
    "projectId": project_id  # ID of the project to liquidate
}
# Send a POST request to the /live/update/liquidate endpoint to liquidate
response = post(f'{BASE_URL}/live/update/liquidate', headers=get_headers(), json=payload)
# Parse the JSON response into python managable dict
result = response.json()
# Check if the request was successful and print the result
if result['success']:
    print("Live Algorithm Liquidated Successfully:")
    print(result)

### Stop Live Algorithm
# Prepare data payload with project ID to stop the algorithm
payload = {
    "projectId": project_id  # ID of the project to stop
}
# Send a POST request to the /live/update/stop endpoint to stop the algorithm
response = post(f'{BASE_URL}/live/update/stop', headers=get_headers(), json=payload)
# Parse the JSON response into python managable dict
result = response.json()
# Check if the request was successful and print the result
if result['success']:
    print("Live Algorithm Stopped Successfully:")
    print(result)

### List Live Algorithms
# Prepare data payload with filters for listing live algorithms
payload = {
    "status": "Running",  # Filter to show only running algorithms
    "start": 1717801200,  # Start time (Unix timestamp) for the list range
    "end": 1743462000  # End time (Unix timestamp) for the list range
}
# Send a POST request to the /live/list endpoint to list algorithms
response = post(f'{BASE_URL}/live/list', headers=get_headers(), json=payload)
# Parse the JSON response into python managable dict
result = response.json()
# Check if the request was successful and print the list
if result['success']:
    print("List of Live Algorithms:")
    print(result)