将数据摄取到 Timescale Cloud 服务中

virtualenv env
source env/bin/activate

复制

pip install twelvedata websocket-client

复制

pip install psycopg2-binary

复制

import time
   from twelvedata import TDClient

    messages_history = []

    def on_event(event):
     print(event) # prints out the data record (dictionary)
     messages_history.append(event)

   td = TDClient(apikey="<YOUR_API_KEY>")
   ws = td.websocket(symbols=["BTC/USD", "ETH/USD"], on_event=on_event)
   ws.subscribe(['ETH/BTC', 'AAPL'])
   ws.connect()
   while True:
   print('messages received: ', len(messages_history))
   ws.heartbeat()
   time.sleep(10)

复制

python websocket_test.py

复制

{'event': 'subscribe-status',
 'status': 'ok',
 'success': [
        {'symbol': 'BTC/USD', 'exchange': 'Coinbase Pro', 'mic_code': 'Coinbase Pro', 'country': '', 'type': 'Digital Currency'},
        {'symbol': 'ETH/USD', 'exchange': 'Huobi', 'mic_code': 'Huobi', 'country': '', 'type': 'Digital Currency'}
    ],
 'fails': None
}

复制

{'event': 'price', 'symbol': 'BTC/USD', 'currency_base': 'Bitcoin', 'currency_quote': 'US Dollar', 'exchange': 'Coinbase Pro', 'type': 'Digital Currency', 'timestamp': 1652438893, 'price': 30361.2, 'bid': 30361.2, 'ask': 30361.2, 'day_volume': 49153}
{'event': 'price', 'symbol': 'BTC/USD', 'currency_base': 'Bitcoin', 'currency_quote': 'US Dollar', 'exchange': 'Coinbase Pro', 'type': 'Digital Currency', 'timestamp': 1652438896, 'price': 30380.6, 'bid': 30380.6, 'ask': 30380.6, 'day_volume': 49157}
{'event': 'heartbeat', 'status': 'ok'}
{'event': 'price', 'symbol': 'ETH/USD', 'currency_base': 'Ethereum', 'currency_quote': 'US Dollar', 'exchange': 'Huobi', 'type': 'Digital Currency', 'timestamp': 1652438899, 'price': 2089.07, 'bid': 2089.02, 'ask': 2089.03, 'day_volume': 193818}
{'event': 'price', 'symbol': 'BTC/USD', 'currency_base': 'Bitcoin', 'currency_quote': 'US Dollar', 'exchange': 'Coinbase Pro', 'type': 'Digital Currency', 'timestamp': 1652438900, 'price': 30346.0, 'bid': 30346.0, 'ask': 30346.0, 'day_volume': 49167}

复制

CREATE TABLE crypto_ticks (
    "time" TIMESTAMPTZ,
    symbol TEXT,
    price DOUBLE PRECISION,
    day_volume NUMERIC
) WITH (
   tsdb.hypertable,
   tsdb.partition_column='time',
   tsdb.segmentby='symbol', 
   tsdb.orderby='time DESC'
);

复制

CREATE TABLE crypto_assets (
    symbol TEXT UNIQUE,
    "name" TEXT
);

复制

import time
import psycopg2

from twelvedata import TDClient
from psycopg2.extras import execute_values
from datetime import datetime

class WebsocketPipeline():
    # name of the hypertable
    DB_TABLE = "stocks_real_time"

    # columns in the hypertable in the correct order
    DB_COLUMNS=["time", "symbol", "price", "day_volume"]

    # batch size used to insert data in batches
    MAX_BATCH_SIZE=100

    def __init__(self, conn):
        """Connect to the Twelve Data web socket server and stream
        data into the database.

        Args:
            conn: psycopg2 connection object
        """
        self.conn = conn
        self.current_batch = []
        self.insert_counter = 0

    def _insert_values(self, data):
        if self.conn is not None:
            cursor = self.conn.cursor()
            sql = f"""
            INSERT INTO {self.DB_TABLE} ({','.join(self.DB_COLUMNS)})
            VALUES %s;"""
            execute_values(cursor, sql, data)
            self.conn.commit()

    def _on_event(self, event):
        """This function gets called whenever there's a new data record coming
        back from the server.

        Args:
            event (dict): data record
        """
        if event["event"] == "price":
            # data record
            timestamp = datetime.utcfromtimestamp(event["timestamp"])
            data = (timestamp, event["symbol"], event["price"], event.get("day_volume"))

            # add new data record to batch
            self.current_batch.append(data)
            print(f"Current batch size: {len(self.current_batch)}")

            # ingest data if max batch size is reached then reset the batch
            if len(self.current_batch) == self.MAX_BATCH_SIZE:
                self._insert_values(self.current_batch)
                self.insert_counter += 1
                print(f"Batch insert #{self.insert_counter}")
                self.current_batch = []
        def start(self, symbols):
            """Connect to the web socket server and start streaming real-time data
            into the database.

            Args:
                symbols (list of symbols): List of stock/crypto symbols
            """
            td = TDClient(apikey="<YOUR_API_KEY")
            ws = td.websocket(on_event=self._on_event)
            ws.subscribe(symbols)
            ws.connect()
            while True:
               ws.heartbeat()
               time.sleep(10)
    onn = psycopg2.connect(database="tsdb",
                        host="<HOST>",
                        user="tsdbadmin",
                        password="<PASSWORD>",
                        port="<PORT>")

    symbols = ["BTC/USD", "ETH/USD", "MSFT", "AAPL"]
    websocket = WebsocketPipeline(conn)
    websocket.start(symbols=symbols)
    ```

复制

python websocket_test.py

复制