リアルタイム データを使用した ETL プロセスの理解: 抽出、変換、読み込み、視覚化

 with websockets.connect(url) as ws:
    response = await ws.recv()
    trade_data = json.loads(response)

price = float(trade_data['p'])
quantity = float(trade_data['q'])
timestamp = int(trade_data['T'])

trade_time = datetime.fromtimestamp(timestamp / 1000.0)

def save_trade_to_db(price, quantity, timestamp):
    conn = sqlite3.connect('trades.db')
    cursor = conn.cursor()
    # Create a table if it doesn't exist
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS trades (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            price REAL,
            quantity REAL,
            timestamp TEXT
        )
    ''')
    # Insert the trade data
    cursor.execute('''
        INSERT INTO trades (price, quantity, timestamp)
        VALUES (?, ?, ?)
    ''', (price, quantity, trade_time))
    conn.commit()
    conn.close()

def plot_trades():
    if len(trades) > 0:
        timestamps, prices, quantities = zip(*trades)

        plt.subplot(2, 1, 1)
        plt.cla()  # Clear the previous plot for real-time updates
        plt.plot(timestamps, prices, label='Price', color='blue')
        plt.ylabel('Price (USDT)')
        plt.legend()
        plt.title('Real-Time BTC/USDT Prices')
        plt.xticks(rotation=45)

        plt.subplot(2, 1, 2)
        plt.cla()  # Clear the previous plot for real-time updates
        plt.plot(timestamps, quantities, label='Quantity', color='orange')
        plt.ylabel('Quantity')
        plt.xlabel('Time')
        plt.legend()
        plt.xticks(rotation=45)

        plt.tight_layout()  # Adjust layout for better spacing
        plt.pause(0.1)  # Pause to update the plot