Comment surveiller l'utilisation et les coûts OpenAI avec l'API Usage et l'API Cost

!pip install requests pandas numpy matplotlib --quiet

import requests
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import time
import json

OPENAI_ADMIN_KEY = '<PLACEHOLDER>'  # Remplacez par votre clé API ADMIN réelle
 
headers = {
    "Authorization": f"Bearer {OPENAI_ADMIN_KEY}",
    "Content-Type": "application/json"
}
 
url = "https://api.openai.com/v1/organization/usage/completions"
 
days_ago = 30
start_time = int(time.time()) - (days_ago * 24 * 60 * 60)
 
params = {
    "start_time": start_time,
    "bucket_width": "1d",
    "limit": 7
}

all_data = []
page_cursor = None
 
while True:
    if page_cursor:
        params["page"] = page_cursor
 
    response = requests.get(url, headers=headers, params=params)
 
    if response.status_code == 200:
        data_json = response.json()
        all_data.extend(data_json.get("data", []))
 
        page_cursor = data_json.get("next_page")
        if not page_cursor:
            break
    else:
        print(f"Erreur : {response.status_code}")
        break
 
if all_data:
    print("Données récupérées avec succès !")
else:
    print("Problème : Aucune donnée disponible à récupérer.")

records = []
 
for bucket in all_data:
    start_time = bucket.get("start_time")
    end_time = bucket.get("end_time")
    for result in bucket.get("results", []):
        records.append({
            "start_time": start_time,
            "end_time": end_time,
            "input_tokens": result.get("input_tokens", 0),
            "output_tokens": result.get("output_tokens", 0),
            "num_model_requests": result.get("num_model_requests", 0)
        })
 
df = pd.DataFrame(records)
df['start_datetime'] = pd.to_datetime(df['start_time'], unit='s')
df['end_datetime'] = pd.to_datetime(df['end_time'], unit='s')
 
df.head()

if not df.empty:
    plt.figure(figsize=(12, 6))
 
    width = 0.35
    indices = range(len(df))
 
    plt.bar(indices, df['input_tokens'], width=width, label='Tokens d\'entrée', alpha=0.7)
    plt.bar([i + width for i in indices], df['output_tokens'], width=width, label='Tokens de sortie', alpha=0.7)
 
    plt.xlabel('Période')
    plt.ylabel('Nombre de tokens')
    plt.title('Utilisation quotidienne des tokens d\'entrée vs sortie - 30 derniers jours')
    plt.xticks([i + width/2 for i in indices], [dt.strftime('%Y-%m-%d') for dt in df['start_datetime']], rotation=45)
    plt.legend()
    plt.tight_layout()
    plt.show()
else:
    print("Aucune donnée disponible pour afficher.")

costs_url = "https://api.openai.com/v1/organization/costs"
all_costs_data = []
page_cursor = None
 
while True:
    costs_params = {
        "start_time": start_time,
        "bucket_width": "1d",
        "limit": 30
    }
 
    if page_cursor:
        costs_params["page"] = page_cursor
 
    costs_response = requests.get(costs_url, headers=headers, params=costs_params)
 
    if costs_response.status_code == 200:
        costs_json = costs_response.json()
        all_costs_data.extend(costs_json.get("data", []))
 
        page_cursor = costs_json.get("next_page")
        if not page_cursor:
            break
    else:
        print(f"Erreur : {costs_response.status_code}")
        break
 
if all_costs_data:
    print("Données de coûts récupérées avec succès !")
else:
    print("Aucune donnée de coûts trouvée.")

cost_records = []
 
for bucket in all_costs_data:
    start_time = bucket.get("start_time")
    end_time = bucket.get("end_time")
    for result in bucket.get("results", []):
        cost_records.append({
            "start_time": start_time,
            "end_time": end_time,
            "amount_value": result.get("amount", {}).get("value", 0),
            "currency": result.get("amount", {}).get("currency", "usd"),
            "line_item": result.get("line_item")
        })
 
cost_df = pd.DataFrame(cost_records)
cost_df['start_datetime'] = pd.to_datetime(cost_df['start_time'], unit='s')
cost_df['end_datetime'] = pd.to_datetime(cost_df['end_time'], unit='s')
 
cost_df.head()

if not cost_df.empty:
    cost_df['date'] = cost_df['start_datetime'].dt.date
    cost_per_day = cost_df.groupby('date')['amount_value'].sum().reset_index()
 
    plt.figure(figsize=(12, 6))
    plt.bar(cost_per_day['date'], cost_per_day['amount_value'], width=0.6, color='skyblue', alpha=0.8)
    plt.xlabel('Date')
    plt.ylabel('Coût total (USD)')
    plt.title('Coût total par jour (30 derniers jours)')
    plt.xticks(rotation=45, ha='right')
    plt.tight_layout()
    plt.show()
else:
    print("Aucune donnée de coûts disponible pour afficher.")