2023 鐵人賽 DAY-16 Shioaji 訂閱即時資料『四』

message FutureRealTimeTickMessage {
    string code              = 1;
    string date_time         = 2;
    double open              = 3;
    double underlying_price  = 4;
    int64 bid_side_total_vol = 5;
    int64 ask_side_total_vol = 6;
    double avg_price         = 7;
    double close             = 8;
    double high              = 9;
    double low               = 10;
    double amount            = 11;
    double total_amount      = 12;
    int64 volume             = 13;
    int64 total_volume       = 14;
    int64 tick_type          = 15;
    int64 chg_type           = 16;
    double price_chg         = 17;
    double pct_chg           = 18;
    bool simtrade            = 19;
}

import threading
import pika

from queue import Queue

from pika.channel import Channel

class PikaCC:
    def __init__(self, conn: pika.BlockingConnection, channel: Channel):
        self.conn = conn
        self.channel = channel

    def heartbeat(self):
        self.conn.process_data_events()


class RabbitMQS:
    def __init__(self, url: str, exchange: str, pool_size: int):
        self.parameters = pika.URLParameters(url)
        self.exchange = exchange
        self.pool_size = pool_size

        # rabbit mq connection queue
        self.pika_queue: Queue = Queue()

        # initial connections
        self.fill_pika_queue()

        # lock
        self.order_cb_lock = threading.Lock()

    def create_pika(self):
        conn = pika.BlockingConnection(self.parameters)
        channel = conn.channel()
        channel.exchange_declare(exchange=self.exchange, exchange_type="direct", durable=True)
        return PikaCC(conn, channel)

    def fill_pika_queue(self):
        for _ in range(self.pool_size):
            self.pika_queue.put(self.create_pika())

def future_quote_callback_v1(self, _, tick: sj.TickFOPv1):
    rabbit = self.pika_queue.get(block=True)
    try:
        rabbit.channel.basic_publish(
            exchange=self.exchange,
            routing_key=f"future_tick:{tick.code}",
            body=mq_pb2.FutureRealTimeTickMessage(
                code=tick.code,
                date_time=datetime.strftime(tick.datetime, "%Y-%m-%d %H:%M:%S.%f"),
                open=tick.open,
                underlying_price=tick.underlying_price,
                bid_side_total_vol=tick.bid_side_total_vol,
                ask_side_total_vol=tick.ask_side_total_vol,
                avg_price=tick.avg_price,
                close=tick.close,
                high=tick.high,
                low=tick.low,
                amount=tick.amount,
                total_amount=tick.total_amount,
                volume=tick.volume,
                total_volume=tick.total_volume,
                tick_type=tick.tick_type,
                chg_type=tick.chg_type,
                price_chg=tick.price_chg,
                pct_chg=tick.pct_chg,
                simtrade=tick.simtrade,
            ).SerializeToString(),
        )
    except Exception as err:
        logger.error("future_quote_callback_v1 error %s", err)
    self.pika_queue.put(rabbit)

class Shioaji:
    def __init__(self):
        self.__api = sj.Shioaji()
        self.__login_status_lock = threading.Lock()
        self.__login_progess = int()

        self.stock_map: dict[str, Contract] = {}
        self.stock_map_lock = threading.Lock()
        self.future_map: dict[str, Contract] = {}
        self.future_map_lock = threading.Lock()

        env = RequiredEnv()
        rabbit = RabbitMQS(
            env.rabbitmq_url,
            env.rabbitmq_exchange,
            64,
        )
        self.rabbit = rabbit