From e50e21fac3a6727e85ef5066bf1c89c27a5128bf Mon Sep 17 00:00:00 2001 From: msincenselee Date: Fri, 11 Dec 2020 15:58:35 +0800 Subject: [PATCH] =?UTF-8?q?[=E5=A2=9E=E5=BC=BA=E5=8A=9F=E8=83=BD]=20?= =?UTF-8?q?=E8=82=A1=E7=A5=A8tick=E8=A1=8C=E6=83=85=E8=AE=A2=E9=98=85?= =?UTF-8?q?=E3=80=81=E6=8E=A8=E9=80=81=E6=9C=8D=E5=8A=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- vnpy/app/stock_tick_publisher/__init__.py | 15 + vnpy/app/stock_tick_publisher/engine.py | 500 ++++++++++++++++++++++ 2 files changed, 515 insertions(+) create mode 100644 vnpy/app/stock_tick_publisher/__init__.py create mode 100644 vnpy/app/stock_tick_publisher/engine.py diff --git a/vnpy/app/stock_tick_publisher/__init__.py b/vnpy/app/stock_tick_publisher/__init__.py new file mode 100644 index 00000000..860f6634 --- /dev/null +++ b/vnpy/app/stock_tick_publisher/__init__.py @@ -0,0 +1,15 @@ +# encoding: UTF-8 + +import os +from pathlib import Path +from vnpy.trader.app import BaseApp +from .engine import StockTickPublisher, APP_NAME + + +class IndexTickPublisherApp(BaseApp): + """""" + app_name = APP_NAME + app_module = __module__ + app_path = Path(__file__).parent + display_name = u'股票tick行情推送' + engine_class = StockTickPublisher diff --git a/vnpy/app/stock_tick_publisher/engine.py b/vnpy/app/stock_tick_publisher/engine.py new file mode 100644 index 00000000..143a2cbe --- /dev/null +++ b/vnpy/app/stock_tick_publisher/engine.py @@ -0,0 +1,500 @@ +# encoding: UTF-8 + +# 股票行情订阅/发布器 +# 华富资产 +import sys +import copy +import json +import traceback +from queue import Queue +from threading import Thread +from datetime import datetime, timedelta +from time import sleep +from logging import ERROR +from typing import Dict + +from vnpy.event import EventEngine +from vnpy.trader.constant import Exchange +from vnpy.trader.engine import BaseEngine, MainEngine +from vnpy.trader.event import EVENT_TIMER +from vnpy.trader.object import TickData +from vnpy.api.rest.rest_client import RestClient +from vnpy.trader.utility import load_json, save_json +from vnpy.amqp.producer import publisher +from vnpy.amqp.consumer import worker + +APP_NAME = 'Stock_Publisher' +REST_HOST = 'http://49.234.35.135:8005' +SUBSCRIBE_FILE = 'today_subscribe.json' + +# 市场交易代码 => vnpy +EXCHANGE_CODE2VT: Dict[str, Exchange] = { + "USHI": Exchange.SSE, # 沪市指数 + "USHA": Exchange.SSE, # 沪市 A 股 + "USHB": Exchange.SSE, # 沪市 B 股 + "USHD": Exchange.SSE, # 沪市债券 + "USHJ": Exchange.SSE, # 沪市基金 + "USHT": Exchange.SSE, # 沪市风险 + "USZI": Exchange.SZSE, # 深市指数 + "USZA": Exchange.SZSE, # 深市 A 股 + "USZB": Exchange.SZSE, # 深市 B 股 + "USZD": Exchange.SZSE, # 深市债券 + "USZJ": Exchange.SZSE, # 深市基金 +} + + +class SubscribeWorker(worker): + """ + 订阅任务执行者 + 它处理 subscribe task queue, + """ + + def __init__(self, parent, gateway_name='', host='localhost', port=5672, user='admin', password='admin', + exchange='x_work_queue', queue='subscribe_task_queue', routing_key='stock_subscribe'): + + self.parent = parent + self.gateway_name = gateway_name + self.routing_key = routing_key + self.start_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S') + + super().__init__(host=host, port=port, user=user, password=password, exchange=exchange, queue=queue, + routing_key=routing_key) + self.thread = Thread(target=self.start) + self.thread.start() + + def callback(self, chan, method_frame, _header_frame, body, userdata=None): + """ + 处理任务的回调函数 + :param chan: + :param method_frame: + :param _header_frame: + :param body: + :param userdata: + :return: + """ + try: + mission = body.decode('utf-8') + self.parent.write_log(" [SubscribeWorker] received task: {}".format(mission)) + + mission = json.loads(mission) + + action = mission.get('action') + if action == 'subscribe': + self.task_subscribe(mission) + + # 订阅任务执行完毕 + chan.basic_ack(delivery_tag=method_frame.delivery_tag) + self.parent.write_log(" [SubscribeWorker] task finished ") + except Exception as ex: + self.parent.write_error('SubscribeWorker Exception:{}'.format(str(ex))) + # self.parent.write_error(traceback.format_exc()) + chan.basic_ack(delivery_tag=method_frame.delivery_tag) + self.parent.write_log(" [SubscribeWorker] task fail ") + + def task_subscribe(self, setting): + """执行订阅任务""" + self.parent.write_log(f'准备添加行情订阅:{setting}') + + self.parent.subscribe(setting.get('vt_symbol')) + return True + + def stop(self): + """ + 停止运行 + :return: + """ + self.parent.write_log(u'worker停止') + try: + self.channel.stop_comsuming() + self.connection.close() + if self.thread: + self.thread.join() + except: + pass + + +class StockRestClient(RestClient): + + def __init__(self, parent=None): + """ + 构造函数 + """ + super().__init__() + self.parent = parent + self.init(url_base=REST_HOST) + + def write_log(self, msg): + """日志""" + if self.parent and hasattr(self.parent, 'write_log'): + func = getattr(self.parent, 'write_log') + func(msg) + else: + print(msg) + + def write_error(self, msg): + """错误日志""" + if self.parent and hasattr(self.parent, 'write_error'): + func = getattr(self.parent, 'write_error') + func(msg) + else: + print(msg, file=sys.stderr) + + +class StockTickPublisher(BaseEngine): + # 股票tick发布服务 + # 接收RabbitMQ接受股票行情订阅请求 + # 通过URL的免费行情接口，获取股票行情tick，发布至rabbitMQ + + # ---------------------------------------------------------------------- + def __init__(self, main_engine: MainEngine, event_engine: EventEngine): + """""" + super().__init__( + main_engine, event_engine, APP_NAME) + + self.main_engine = main_engine + self.event_engine = event_engine + self.create_logger(logger_name=APP_NAME) + + self.last_minute = None + + self.register_event() + + self.req_interval = 1 # 操作请求间隔1秒 + self.req_id = 0 # 操作请求编号 + self.connection_status = False # 连接状态 + + self.api = None # API 的连接会话对象 + self.last_tick_dt = None # 记录该会话对象的最后一个tick时间 + + self.subscribe_symbols = set() + + # 获取持久化订阅得合约 + self.subscribe_dict = load_json(SUBSCRIBE_FILE, auto_save=True) + cur_trading_day = datetime.now().strftime('%Y-%m-%d') + # 对当前交易日得订阅合约，逐一添加 + for symbol in self.subscribe_dict.get(cur_trading_day, []): + self.write_log(f'添加订阅{cur_trading_day}的股票:{symbol}') + self.subscribe_symbols.add(symbol) + # 排除非当前交易日点阅得合约 + for trading_day in list(self.subscribe_dict.keys()): + if trading_day != cur_trading_day: + self.subscribe_dict.pop(trading_day, None) + + # 重新保存 + save_json(SUBSCRIBE_FILE, self.subscribe_dict) + + # vt_setting.json内rabbitmq配置项 + self.conf = {} + self.pub = None # 行情推送器 + self.worker = None # 订阅任务处理 + self.req_thread = None + + # 负责执行数据库插入的单独线程相关 + # 是否启动 + self.active = False + self.pub_queue = Queue() # 队列 + self.pub_thread = None + + def write_error(self, content: str): + self.write_log(msg=content, level=ERROR) + + def create_publisher(self, conf): + """创建rabbitmq 消息发布器""" + self.write_log(f'创建rabbitmq 消息发布器') + if self.pub: + return + host = conf.get('host', 'localhost') + port = conf.get('port', 5672) + user = conf.get('user', 'admin') + password = conf.get('password', 'admin') + channel_number = conf.get('channel_number', 1) + queue_name = conf.get('queue_name', '') + routing_key = conf.get('routing_key', 'default') + exchange = conf.get('exchange', 'x_fanout_stock_tick') + + self.write_log(f'创建rabbit MQ消息发布器:{host}:{port}/ex:{exchange}/q_n:{queue_name}/r_k:{routing_key}') + try: + # 消息发布 + self.pub = publisher(host=host, + port=port, + user=user, + password=password, + channel_number=channel_number, + queue_name=queue_name, + routing_key=routing_key, + exchange=exchange) + + except Exception as ex: + self.write_error(u'创建tick发布器异常:{}'.format(str(ex))) + + def create_worker(self, conf): + """ + 创建订阅任务接收器 + :param conf: + :return: + """ + host = conf.get('host', 'localhost') + port = conf.get('port', 5672) + user = conf.get('user', 'admin') + password = conf.get('password', 'admin') + exchange = 'x_work_queue' + queue = 'subscribe_task_queue' + routing_key = 'stock_subscribe' + self.write_error(f'创建订阅任务接收器:{host}:{port}/q_n:{queue}/r_k:{routing_key}') + try: + self.worker = SubscribeWorker( + parent=self, + host=host, + port=port, + user=user, + password=password, + exchange=exchange, + queue=queue, + routing_key=routing_key) + self.worker.start() + except Exception as ex: + self.write_error(f'创建订阅任务接收器异常:{str(ex)}') + + # ---------------------------------------------------------------------- + def register_event(self): + """注册事件监听""" + self.event_engine.register(EVENT_TIMER, self.process_timer_event) + + def process_timer_event(self, event): + """定时执行""" + dt = datetime.now() + + if dt.minute == self.last_minute: + return + + def connect(self, rabbit_config: dict): + """ + 连接股票行情服务器 + :param n: + :return: + """ + if self.connection_status: + if self.api is not None or getattr(self.api, "client", None) is not None: + self.write_log(u'当前已经连接,不需要重新连接') + return + + self.write_log(u'开始接入股票行情服务器') + try: + self.write_log(f'创建Rest客户端') + self.api = StockRestClient() + self.write_log(f'启动Rest客户端服务') + self.api.start() + self.connection_status = True + + except Exception as ex: + self.write_error(u'连接Rest服务器异常:{},{}'.format(str(ex), traceback.format_exc())) + + return + + # 更新配置 + self.conf.update(rabbit_config) + + # 创建推送线程 + self.pub_thread = Thread(target=self.run_pub) # 线程 + + self.pub_thread.start() + + # 创建请求线程 + self.req_thread = Thread(target=self.run_req) + + # 启动请求线程 + self.req_thread.start() + + # 创建处理订阅线程 + self.create_worker(self.conf) + + def subscribe(self, symbol): + """订阅股票行情""" + self.write_log(f'订阅股票行情:{symbol}') + if '.' in symbol: + symbol = symbol.split('.')[0] + self.write_log(f'合约=>{symbol}') + # 添加订阅 + self.subscribe_symbols.add(symbol) + + # 持久化=》json文件 + today = datetime.now().strftime('%Y-%m-%d') + subscribe_symbols = self.subscribe_dict.get(today, []) + if symbol not in subscribe_symbols: + subscribe_symbols.append(symbol) + self.subscribe_dict[today] = subscribe_symbols + save_json(SUBSCRIBE_FILE, self.subscribe_dict) + + def close(self): + """退出API""" + self.write_log(u'退出Rest API') + self.connection_status = False + self.active = False + if self.api: + self.write_log(f'退出Rest客户端') + self.api.stop() + self.api.join() + + if self.req_thread is not None: + self.write_log(u'退出请求线程') + self.req_thread.join() + + if self.pub: + self.write_log(u'退出rabbitMQ 发布器') + self.pub.exit() + + if self.pub_thread is not None: + self.write_log(f'退出pub线程') + self.pub_thread.join() + + def run_req(self): + + try: + last_dt = datetime.now() + self.write_log(u'开始运行行情轮询,{}'.format(last_dt)) + while self.connection_status: + try: + self.query_tick( + symbols=list(self.subscribe_symbols) + ) + + except Exception as ex: + self.write_error(u'rest exception:{},{}'.format(str(ex), traceback.format_exc())) + + sleep(self.req_interval) + dt = datetime.now() + if last_dt.minute != dt.minute: + self.write_log('check point. {},last_tick_dt:{}'.format(dt, self.last_tick_dt)) + last_dt = dt + except Exception as ex: + self.write_error(u'rest pool.run exception:{},{}'.format(str(ex), traceback.format_exc())) + + self.write_error(u'rest 线程 {}退出'.format(datetime.now())) + + # ---------------------------------------------------------------------- + def run_pub(self): + """运行推送线程""" + try: + self.create_publisher(self.conf) + self.active = True + self.write_log(f'启动推送线程') + while self.active: + try: + d = self.pub_queue.get(block=True, timeout=1) + if self.pub: + self.pub.pub(d) + except Exception as ex: # noqa + pass + except Exchange as ex: + pass + self.write_log(f'推送线程结束') + + def check_error(self, data: dict, func: str = ""): + """""" + if data["err_code"] == 0: + return False + + error_msg = data["err_msg"] + + self.write_log(f"{func}请求出错，信息：{error_msg}") + return True + + def query_tick(self, symbols): + """ + 查询基础行情数据，不包括买卖盘数据 + :param symbols: + :return: + """ + total_symbols = len(symbols) + if total_symbols == 0: + return + # 每次请求，不超过10个股票 + for i in range(int(total_symbols / 10) + 1): + path = "/stock/query/comm?code={}".format(','.join(symbols[i * 10:10 * (i + 1)])) + self.api.add_request( + method="GET", + path=path, + callback=self.on_query_tick + ) + + def on_query_tick(self, data, request): + """处理股票行情tick""" + + if self.check_error(data, "查询股票实时行情"): + return + + for d in data.get('data', []): + symbol = d.get('code', None) + if symbol is None or len(symbol) != 6: + continue + market = d.get('market') + exchange = EXCHANGE_CODE2VT.get(market, None) + if exchange is None: + continue + + tick_datetime = datetime.now() + + # # 修正毫秒 + # last_tick = self.symbol_tick_dict.get(vn_symbol, None) + # if (last_tick is not None) and tick_datetime.replace(microsecond=0) == last_tick.datetime: + # # 与上一个tick的时间（去除毫秒后）相同,修改为500毫秒 + # tick_datetime = tick_datetime.replace(microsecond=500) + # else: + tick_datetime = tick_datetime.replace(microsecond=0) + + tick = TickData( + gateway_name='tdx', + symbol=symbol, + datetime=tick_datetime, + exchange=exchange + ) + + tick.pre_close = float(d.get('prev_close', 0.0)) + tick.high_price = float(d.get('high', 0.0)) + tick.open_price = float(d.get('open', 0.0)) + tick.low_price = float(d.get('low', 0.0)) + tick.last_price = float(d.get('price', 0.0)) + + tick.volume = int(d.get('tran_volume', 0)) + tick.open_interest = d.get('tran_amount') + + tick.time = tick.datetime.strftime('%H:%M:%S.%f')[0:12] + tick.date = tick.datetime.strftime('%Y-%m-%d') + + tick.trading_day = tick.date + + # 指数没有涨停和跌停，就用昨日收盘价正负10% + tick.limit_up = float(d.get('high_limit')) + tick.limit_down = float(d.get('low_limit')) + + # 只有一档行情 + tick.bid_price_1 = float(d.get('B1', 0.0)) + tick.bid_volume_1 = int(d.get('B1V', 0)) + tick.ask_price_1 = float(d.get('S1', 0.0)) + tick.ask_volume_1 = int(d.get('S1V', 0)) + + # 排除非交易时间得tick + + if tick.datetime.hour not in [9, 10, 11, 13, 14, 15]: + continue + if tick.datetime.hour == 9 and tick.datetime.minute < 30: + continue + # 排除早盘 11:30~12:00 + if tick.datetime.hour == 11 and tick.datetime.minute >= 30: + continue + if tick.datetime.hour == 15 and tick.datetime.minute >= 1: + continue + + self.last_tick_dt = tick.datetime + + # self.symbol_tick_dict[tick.symbol] = tick + # =》写入本地队列 + d = copy.copy(tick.__dict__) + if isinstance(tick.datetime, datetime): + d.update({'datetime': tick.datetime.strftime('%Y-%m-%d %H:%M:%S.%f')}) + d.update({'exchange': tick.exchange.value}) + d = json.dumps(d) + self.pub_queue.put(d) + + #if self.pub: + # self.pub.pub(d)