senset/vnpy
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

[增强功能] CTA引擎运行时重加载新策略代码，融航Gateway，Position增加名称,tdx支持socks5代理

Browse Source
This commit is contained in:
msincenselee 2020-05-22 15:54:05 +08:00
parent 83d2a40624
commit 63d9ea4da7
19 changed files with 1310 additions and 98 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
vnpy/amqp/consumer.py
View File

@ -32,7 +32,7 @@ class receiver(base_broker):
# self.channel.basic_qos(prefetch_count=1) # self.channel.basic_qos(prefetch_count=1)
def callback(self, chan, method_frame, _header_frame, body, userdata=None): def callback(self, chan, method_frame, _header_frame, body, userdata=None):
print(1) #print(1)
print(" [x] received: %r" % body) print(" [x] received: %r" % body)
def subscribe(self): def subscribe(self):
@ -44,7 +44,7 @@ class receiver(base_broker):
try: try:
self.subscribe() self.subscribe()
except Exception as e: except Exception as e:
print(e) print('start consumer exception:{}'.format(str(e)))
self.start() self.start()
@ -72,7 +72,7 @@ class worker(base_broker):
self.channel.basic_qos(prefetch_count=1) self.channel.basic_qos(prefetch_count=1)
def callback(self, chan, method_frame, _header_frame, body, userdata=None): def callback(self, chan, method_frame, _header_frame, body, userdata=None):
print(1) #print(1)
print(" [x] received task: %r" % body) print(" [x] received task: %r" % body)
chan.basic_ack(delivery_tag=method_frame.delivery_tag) chan.basic_ack(delivery_tag=method_frame.delivery_tag)
print(" [x] task finished ") print(" [x] task finished ")
@ -120,7 +120,7 @@ class subscriber(base_broker):
self.cb_func = cb_func self.cb_func = cb_func
def callback(self, chan, method_frame, _header_frame, body, userdata=None): def callback(self, chan, method_frame, _header_frame, body, userdata=None):
print(1) #print(1)
print(" [x] %r" % body) print(" [x] %r" % body)
def subscribe(self): def subscribe(self):
@ -160,7 +160,7 @@ class subscriber_routing(base_broker):
routing_key=routing_key) routing_key=routing_key)
def callback(self, chan, method_frame, _header_frame, body, userdata=None): def callback(self, chan, method_frame, _header_frame, body, userdata=None):
print(1) #print(1)
print(" [x] %r" % body) print(" [x] %r" % body)
def subscribe(self): def subscribe(self):
@ -198,7 +198,7 @@ class subscriber_topic(base_broker):
routing_key=routing_key) routing_key=routing_key)
def callback(self, chan, method_frame, _header_frame, body, userdata=None): def callback(self, chan, method_frame, _header_frame, body, userdata=None):
print(1) #print(1)
print(" [x] %r" % body) print(" [x] %r" % body)
def subscribe(self): def subscribe(self):

23
vnpy/app/account_recorder/engine.py
View File

@ -93,6 +93,8 @@ class AccountRecorder(BaseEngine):
# 账号的同步记录 # 账号的同步记录
self.account_dict = {} # gateway_name: setting self.account_dict = {} # gateway_name: setting
self.is_7x24 = False # 7 x 24 运行的账号( 数字货币)
self.last_qry_dict = {} self.last_qry_dict = {}
self.copy_history_orders = [] # 需要复制至历史成交的gateway名称 self.copy_history_orders = [] # 需要复制至历史成交的gateway名称
self.copy_history_trades = [] # 需要复制至历史成交的gateway名称 self.copy_history_trades = [] # 需要复制至历史成交的gateway名称
@ -101,7 +103,7 @@ class AccountRecorder(BaseEngine):
self.gw_name_acct_id = {} self.gw_name_acct_id = {}
self.is_remove_pre_data = False self.cur_trading_date = ""
self.scaning_gw = [] self.scaning_gw = []
@ -143,6 +145,7 @@ class AccountRecorder(BaseEngine):
# 获取需要处理处理得账号配置 # 获取需要处理处理得账号配置
self.account_dict = d.get('accounts', {}) self.account_dict = d.get('accounts', {})
self.is_7x24 = d.get('is_7x24', False)
# 识别配置，检查账号是否需要复制委托/成交到历史表 # 识别配置，检查账号是否需要复制委托/成交到历史表
for gateway_name, account_setting in self.account_dict.items(): for gateway_name, account_setting in self.account_dict.items():
@ -288,7 +291,7 @@ class AccountRecorder(BaseEngine):
:param trading_day: :param trading_day:
:return: :return:
""" """
if self.is_remove_pre_data: if self.cur_trading_date == trading_day:
return return
# 移除非当日得交易/持仓 # 移除非当日得交易/持仓
@ -322,7 +325,7 @@ class AccountRecorder(BaseEngine):
col_name=TODAY_STRATEGY_POS_COL, col_name=TODAY_STRATEGY_POS_COL,
flt=flt) flt=flt)
self.is_remove_pre_data = True self.cur_trading_date = trading_day
def update_order(self, event: Event): def update_order(self, event: Event):
"""更新当日记录""" """更新当日记录"""
@ -374,10 +377,16 @@ class AccountRecorder(BaseEngine):
self.update_data(db_name=ACCOUNT_DB_NAME, col_name=HISTORY_ORDER_COL, fld=fld2, data=history_data) self.update_data(db_name=ACCOUNT_DB_NAME, col_name=HISTORY_ORDER_COL, fld=fld2, data=history_data)
def get_trading_date(self, dt:datetime):
if self.is_7x24:
return dt.strftime('%Y-%m-%d')
else:
return get_trading_date(dt)
def update_trade(self, event: Event): def update_trade(self, event: Event):
"""更新当日成交""" """更新当日成交"""
trade = event.data trade = event.data
trade_date = get_trading_date(datetime.now()) trade_date = self.get_trading_date(datetime.now())
fld = {'vt_symbol': trade.vt_symbol, fld = {'vt_symbol': trade.vt_symbol,
'account_id': trade.accountid, 'account_id': trade.accountid,
@ -413,7 +422,7 @@ class AccountRecorder(BaseEngine):
def update_position(self, event: Event): def update_position(self, event: Event):
"""更新当日持仓""" """更新当日持仓"""
pos = event.data pos = event.data
trade_date = get_trading_date(datetime.now()) trade_date = self.get_trading_date(datetime.now())
# 不处理交易所返回得套利合约 # 不处理交易所返回得套利合约
if pos.symbol.startswith('SP') and '&' in pos.symbol and ' ' in pos.symbol: if pos.symbol.startswith('SP') and '&' in pos.symbol and ' ' in pos.symbol:
@ -532,7 +541,7 @@ class AccountRecorder(BaseEngine):
d.update({'msg': data.msg}) d.update({'msg': data.msg})
d.update({'additional_info': data.additional_info}) d.update({'additional_info': data.additional_info})
d.update({'log_time': datetime.now().strftime("%Y-%m-%d %H:%M:%S")}) d.update({'log_time': datetime.now().strftime("%Y-%m-%d %H:%M:%S")})
d.update({'trading_day': get_trading_date()}) d.update({'trading_day': self.get_trading_date(datetime.now())})
account_id = self.gw_name_acct_id.get(data.gateway_name, None) account_id = self.gw_name_acct_id.get(data.gateway_name, None)
if account_id: if account_id:
@ -584,7 +593,7 @@ class AccountRecorder(BaseEngine):
d.update({'msg': data.msg}) d.update({'msg': data.msg})
d.update({'additional_info': data.additional_info}) d.update({'additional_info': data.additional_info})
d.update({'log_time': datetime.now().strftime("%Y-%m-%d %H:%M:%S")}) d.update({'log_time': datetime.now().strftime("%Y-%m-%d %H:%M:%S")})
d.update({'trading_day': get_trading_date()}) d.update({'trading_day': self.get_trading_date(datetime.now())})
account_id = self.gw_name_acct_id.get(data.gateway_name, None) account_id = self.gw_name_acct_id.get(data.gateway_name, None)
if account_id: if account_id:

23
vnpy/app/cta_crypto/engine.py
View File

@ -58,7 +58,8 @@ from vnpy.trader.utility import (
TRADER_DIR, TRADER_DIR,
get_folder_path, get_folder_path,
get_underlying_symbol, get_underlying_symbol,
append_data) append_data,
import_module_by_str)
from vnpy.trader.util_logger import setup_logger, logging from vnpy.trader.util_logger import setup_logger, logging
from vnpy.trader.util_wechat import send_wx_msg from vnpy.trader.util_wechat import send_wx_msg
@ -1128,10 +1129,22 @@ class CtaEngine(BaseEngine):
module_name = self.class_module_map[class_name] module_name = self.class_module_map[class_name]
# 重新load class module # 重新load class module
if not self.load_strategy_class_from_module(module_name): #if not self.load_strategy_class_from_module(module_name):
err_msg = f'不能加载模块:{module_name}' # err_msg = f'不能加载模块:{module_name}'
self.write_error(err_msg) # self.write_error(err_msg)
return False, err_msg # return False, err_msg
if module_name:
new_class_name = module_name + '.' + class_name
self.write_log(u'转换策略为全路径:{}'.format(new_class_name))
strategy_class = import_module_by_str(new_class_name)
if strategy_class is None:
err_msg = u'加载策略模块失败:{}'.format(class_name)
self.write_error(err_msg)
return False, err_msg
self.write_log(f'重新加载模块成功，使用新模块:{new_class_name}')
self.classes[class_name] = strategy_class
# 停止当前策略实例的运行，撤单 # 停止当前策略实例的运行，撤单
self.stop_strategy(strategy_name) self.stop_strategy(strategy_name)

26
vnpy/app/cta_stock/engine.py
View File

@ -59,7 +59,8 @@ from vnpy.trader.utility import (
TRADER_DIR, TRADER_DIR,
get_folder_path, get_folder_path,
get_underlying_symbol, get_underlying_symbol,
append_data) append_data,
import_module_by_str)
from vnpy.trader.util_logger import setup_logger, logging from vnpy.trader.util_logger import setup_logger, logging
from vnpy.trader.util_wechat import send_wx_msg from vnpy.trader.util_wechat import send_wx_msg
@ -210,7 +211,7 @@ class CtaEngine(BaseEngine):
all_trading = False all_trading = False
dt = datetime.now() dt = datetime.now()
# 每分钟执行的逻辑
if self.last_minute != dt.minute: if self.last_minute != dt.minute:
self.last_minute = dt.minute self.last_minute = dt.minute
@ -218,6 +219,7 @@ class CtaEngine(BaseEngine):
# 主动获取所有策略得持仓信息 # 主动获取所有策略得持仓信息
all_strategy_pos = self.get_all_strategy_pos() all_strategy_pos = self.get_all_strategy_pos()
# 每5分钟检查一次
if dt.minute % 5 == 0: if dt.minute % 5 == 0:
# 比对仓位，使用上述获取得持仓信息，不用重复获取 # 比对仓位，使用上述获取得持仓信息，不用重复获取
self.compare_pos(strategy_pos_list=copy(all_strategy_pos)) self.compare_pos(strategy_pos_list=copy(all_strategy_pos))
@ -1148,10 +1150,22 @@ class CtaEngine(BaseEngine):
module_name = self.class_module_map[class_name] module_name = self.class_module_map[class_name]
# 重新load class module # 重新load class module
if not self.load_strategy_class_from_module(module_name): #if not self.load_strategy_class_from_module(module_name):
err_msg = f'不能加载模块:{module_name}' # err_msg = f'不能加载模块:{module_name}'
self.write_error(err_msg) # self.write_error(err_msg)
return False, err_msg # return False, err_msg
if module_name:
new_class_name = module_name + '.' + class_name
self.write_log(u'转换策略为全路径:{}'.format(new_class_name))
strategy_class = import_module_by_str(new_class_name)
if strategy_class is None:
err_msg = u'加载策略模块失败:{}'.format(class_name)
self.write_error(err_msg)
return False, err_msg
self.write_log(f'重新加载模块成功，使用新模块:{new_class_name}')
self.classes[class_name] = strategy_class
# 停止当前策略实例的运行，撤单 # 停止当前策略实例的运行，撤单
self.stop_strategy(strategy_name) self.stop_strategy(strategy_name)

4
vnpy/app/cta_stock/template.py
View File

@ -371,7 +371,7 @@ class StockPolicy(CtaPolicy):
super().from_json(json_data) super().from_json(json_data)
self.cur_trading_date = json_data.get('cur_trading_date', None) self.cur_trading_date = json_data.get('cur_trading_date', None)
self.sub_tns = json_data.get('sub_tns') self.sub_tns = json_data.get('sub_tns',{})
signals = json_data.get('signals', {}) signals = json_data.get('signals', {})
for kline_name, signal in signals: for kline_name, signal in signals:
last_signal = signal.get('last_signal', "") last_signal = signal.get('last_signal', "")
@ -872,7 +872,7 @@ class CtaStockTemplate(CtaTemplate):
continue continue
cur_price = self.cta_engine.get_price(lg.vt_symbol) cur_price = self.cta_engine.get_price(lg.vt_symbol)
if not lg.stop_price and lg.stop_price > cur_price > 0: if lg.stop_price != 0 and lg.stop_price > cur_price > 0:
# 调用平仓模块 # 调用平仓模块
self.write_log(u'{} {}当前价:{} 触发止损线{},开仓价:{},v：{}'. self.write_log(u'{} {}当前价:{} 触发止损线{},开仓价:{},v：{}'.
format(self.cur_datetime, format(self.cur_datetime,

32
vnpy/app/cta_strategy_pro/engine.py
View File

@ -55,7 +55,8 @@ from vnpy.trader.utility import (
TRADER_DIR, TRADER_DIR,
get_folder_path, get_folder_path,
get_underlying_symbol, get_underlying_symbol,
append_data) append_data,
import_module_by_str)
from vnpy.trader.util_logger import setup_logger, logging from vnpy.trader.util_logger import setup_logger, logging
from vnpy.trader.util_wechat import send_wx_msg from vnpy.trader.util_wechat import send_wx_msg
@ -351,7 +352,7 @@ class CtaEngine(BaseEngine):
self.put_strategy_event(strategy) self.put_strategy_event(strategy)
if self.engine_config.get('trade_2_wx', False): if self.engine_config.get('trade_2_wx', False):
accountid = self.engine_config.get('accountid', '-') accountid = self.engine_config.get('accountid', 'XXX')
d = { d = {
'account': accountid, 'account': accountid,
'strategy': strategy_name, 'strategy': strategy_name,
@ -362,7 +363,7 @@ class CtaEngine(BaseEngine):
'remark': f'{accountid}:{strategy_name}', 'remark': f'{accountid}:{strategy_name}',
'timestamp': trade.time 'timestamp': trade.time
} }
send_wx_msg(content=d, target=accountid) send_wx_msg(content=d, target=accountid, msg_type='TRADE')
def process_position_event(self, event: Event): def process_position_event(self, event: Event):
"""""" """"""
@ -829,6 +830,9 @@ class CtaEngine(BaseEngine):
return None return None
def get_contract(self, vt_symbol):
return self.main_engine.get_contract(vt_symbol)
def get_account(self, vt_accountid: str = ""): def get_account(self, vt_accountid: str = ""):
""" 查询账号的资金""" """ 查询账号的资金"""
# 如果启动风控，则使用风控中的最大仓位 # 如果启动风控，则使用风控中的最大仓位
@ -1115,10 +1119,22 @@ class CtaEngine(BaseEngine):
module_name = self.class_module_map[class_name] module_name = self.class_module_map[class_name]
# 重新load class module # 重新load class module
if not self.load_strategy_class_from_module(module_name): #if not self.load_strategy_class_from_module(module_name):
err_msg = f'不能加载模块:{module_name}' # err_msg = f'不能加载模块:{module_name}'
self.write_error(err_msg) # self.write_error(err_msg)
return False, err_msg # return False, err_msg
if module_name:
new_class_name = module_name + '.' + class_name
self.write_log(u'转换策略为全路径:{}'.format(new_class_name))
strategy_class = import_module_by_str(new_class_name)
if strategy_class is None:
err_msg = u'加载策略模块失败:{}'.format(new_class_name)
self.write_error(err_msg)
return False, err_msg
self.write_log(f'重新加载模块成功，使用新模块:{new_class_name}')
self.classes[class_name] = strategy_class
# 停止当前策略实例的运行，撤单 # 停止当前策略实例的运行，撤单
self.stop_strategy(strategy_name) self.stop_strategy(strategy_name)
@ -1839,7 +1855,7 @@ class CtaEngine(BaseEngine):
print(f"{strategy_name}: {msg}" if strategy_name else msg, file=sys.stderr) print(f"{strategy_name}: {msg}" if strategy_name else msg, file=sys.stderr)
if level in [logging.CRITICAL, logging.WARN, logging.WARNING]: if level in [logging.CRITICAL, logging.WARN, logging.WARNING]:
send_wx_msg(content=f"{strategy_name}: {msg}" if strategy_name else msg) send_wx_msg(content=f"{strategy_name}: {msg}" if strategy_name else msg, target=self.engine_config.get('accountid', 'XXX'))
def write_error(self, msg: str, strategy_name: str = '', level: int = logging.ERROR): def write_error(self, msg: str, strategy_name: str = '', level: int = logging.ERROR):
"""写入错误日志""" """写入错误日志"""

4
vnpy/app/cta_strategy_pro/template.py
View File

@ -636,7 +636,6 @@ class CtaProTemplate(CtaTemplate):
self.symbol_size = self.cta_engine.get_size(self.vt_symbol) self.symbol_size = self.cta_engine.get_size(self.vt_symbol)
self.margin_rate = self.cta_engine.get_margin_rate(self.vt_symbol) self.margin_rate = self.cta_engine.get_margin_rate(self.vt_symbol)
def sync_data(self): def sync_data(self):
"""同步更新数据""" """同步更新数据"""
if not self.backtesting: if not self.backtesting:
@ -1111,7 +1110,6 @@ class CtaProFutureTemplate(CtaProTemplate):
self.trading = True self.trading = True
self.put_event() self.put_event()
# ----------------------------------------------------------------------
def on_stop(self): def on_stop(self):
"""停止策略（必须由用户继承实现）""" """停止策略（必须由用户继承实现）"""
self.active_orders.clear() self.active_orders.clear()
@ -1290,7 +1288,6 @@ class CtaProFutureTemplate(CtaProTemplate):
order_vt_symbol = copy(old_order['vt_symbol']) order_vt_symbol = copy(old_order['vt_symbol'])
order_volume = old_order['volume'] - old_order['traded'] order_volume = old_order['volume'] - old_order['traded']
order_price = old_order['price'] order_price = old_order['price']
order_type = old_order.get('order_type', OrderType.LIMIT) order_type = old_order.get('order_type', OrderType.LIMIT)
order_retry = old_order.get('retry', 0) order_retry = old_order.get('retry', 0)
@ -1471,7 +1468,6 @@ class CtaProFutureTemplate(CtaProTemplate):
self.active_orders.pop(order.vt_orderid, None) self.active_orders.pop(order.vt_orderid, None)
return return
if order_retry > 20: if order_retry > 20:
msg = u'{} 平仓撤单 {}/{}手， 重试平仓次数{}>20' \ msg = u'{} 平仓撤单 {}/{}手， 重试平仓次数{}>20' \
.format(self.strategy_name, order_vt_symbol, order_volume, order_retry) .format(self.strategy_name, order_vt_symbol, order_volume, order_retry)

3
vnpy/data/tdx/tdx_common.py
View File

@ -27,6 +27,8 @@ TDX_FUTURE_CONFIG = 'tdx_future_config.json'
# } } # } }
TDX_STOCK_CONFIG = 'tdx_stock_config.pkb2' TDX_STOCK_CONFIG = 'tdx_stock_config.pkb2'
TDX_PROXY_CONFIG = 'tdx_proxy_config.json'
@lru_cache() @lru_cache()
def get_tdx_market_code(code): def get_tdx_market_code(code):
@ -94,7 +96,6 @@ def save_future_contracts(future_contracts_dict: dict):
"""保存期货合约信息""" """保存期货合约信息"""
save_cache_json(future_contracts_dict, 'future_contracts.json') save_cache_json(future_contracts_dict, 'future_contracts.json')
def get_cache_config(config_file_name): def get_cache_config(config_file_name):
"""获取本地缓存的配置地址信息""" """获取本地缓存的配置地址信息"""
config_file_name = os.path.abspath(os.path.join(os.path.dirname(__file__), config_file_name)) config_file_name = os.path.abspath(os.path.join(os.path.dirname(__file__), config_file_name))

25
vnpy/data/tdx/tdx_future_data.py
View File

@ -91,11 +91,13 @@ def get_tdx_marketid(symbol):
class TdxFutureData(object): class TdxFutureData(object):
# ---------------------------------------------------------------------- # ----------------------------------------------------------------------
def __init__(self, strategy=None, best_ip={}): def __init__(self, strategy=None, best_ip={}, proxy_ip="", proxy_port=0):
""" """
构造函数 构造函数
:param strategy: 上层策略，主要用与使用write_log（） :param strategy: 上层策略，主要用与使用write_log（）
""" """
self.proxy_ip = proxy_ip
self.proxy_port = proxy_port
self.api = None self.api = None
self.connection_status = False # 连接状态 self.connection_status = False # 连接状态
self.best_ip = best_ip self.best_ip = best_ip
@ -145,7 +147,14 @@ class TdxFutureData(object):
if len(self.best_ip) == 0: if len(self.best_ip) == 0:
self.best_ip = self.select_best_ip() self.best_ip = self.select_best_ip()
self.api.connect(self.best_ip['ip'], self.best_ip['port']) # 如果配置proxy5，使用vnpy项目下的pytdx
if len(self.proxy_ip) > 0 and self.proxy_port > 0:
self.api.connect(ip=self.best_ip['ip'], port=self.best_ip['port'],
proxy_ip=self.proxy_ip, proxy_port=self.proxy_port)
else:
# 使用pip install pytdx
self.api.connect(ip=self.best_ip['ip'], port=self.best_ip['port'])
# 尝试获取市场合约统计 # 尝试获取市场合约统计
c = self.api.get_instrument_count() c = self.api.get_instrument_count()
if c < 10: if c < 10:
@ -176,7 +185,7 @@ class TdxFutureData(object):
apix = TdxExHq_API() apix = TdxExHq_API()
__time1 = datetime.now() __time1 = datetime.now()
try: try:
with apix.connect(ip, port): with apix.connect(ip=ip, port=port, proxy_ip=self.proxy_ip, proxy_port=self.proxy_port):
if apix.get_instrument_count() > 10000: if apix.get_instrument_count() > 10000:
_timestamp = datetime.now() - __time1 _timestamp = datetime.now() - __time1
self.write_log(f'服务器{ip}:{port},耗时:{_timestamp}') self.write_log(f'服务器{ip}:{port},耗时:{_timestamp}')
@ -534,6 +543,8 @@ class TdxFutureData(object):
# 查询的是指数合约 # 查询的是指数合约
symbol = symbol.replace('99', 'L9') symbol = symbol.replace('99', 'L9')
tdx_index_symbol = symbol tdx_index_symbol = symbol
elif symbol.endswith('L9'):
tdx_index_symbol = symbol
else: else:
# 查询的是普通合约 # 查询的是普通合约
tdx_index_symbol = get_underlying_symbol(symbol).upper() + 'L9' tdx_index_symbol = get_underlying_symbol(symbol).upper() + 'L9'
@ -543,8 +554,8 @@ class TdxFutureData(object):
q_size = QSIZE * 5 q_size = QSIZE * 5
# 每秒 2个， 10小时 # 每秒 2个， 10小时
max_data_size = 1000000 max_data_size = 1000000
market_id = INIT_TDX_MARKET_MAP.get(tdx_index_symbol, 0)
self.write_log(u'开始下载{}当日分笔数据'.format(symbol)) self.write_log(u'开始下载{}=>{}, market_id={} 当日分笔数据'.format(symbol,tdx_index_symbol, market_id))
try: try:
_datas = [] _datas = []
@ -552,7 +563,7 @@ class TdxFutureData(object):
while True: while True:
_res = self.api.get_transaction_data( _res = self.api.get_transaction_data(
market=self.symbol_market_dict.get(tdx_index_symbol, 0), market=market_id,
code=symbol, code=symbol,
start=_pos, start=_pos,
count=q_size) count=q_size)
@ -667,6 +678,8 @@ class TdxFutureData(object):
# 查询的是指数合约 # 查询的是指数合约
symbol = symbol.replace('99', 'L9') symbol = symbol.replace('99', 'L9')
tdx_index_symbol = symbol tdx_index_symbol = symbol
elif symbol.endswith('L9'):
tdx_index_symbol = symbol
else: else:
# 查询的是普通合约 # 查询的是普通合约
tdx_index_symbol = get_underlying_symbol(symbol).upper() + 'L9' tdx_index_symbol = get_underlying_symbol(symbol).upper() + 'L9'

9
vnpy/data/tdx/tdx_stock_data.py
View File

@ -120,7 +120,14 @@ class TdxStockData(object):
self.config.update({'best_ip': self.best_ip}) self.config.update({'best_ip': self.best_ip})
save_cache_config(self.config, TDX_STOCK_CONFIG) save_cache_config(self.config, TDX_STOCK_CONFIG)
self.api.connect(self.best_ip.get('ip'), self.best_ip.get('port')) # 如果配置proxy5，使用vnpy项目下的pytdx
if len(self.proxy_ip) > 0 and self.proxy_port > 0:
self.api.connect(ip=self.best_ip['ip'], port=self.best_ip['port'],
proxy_ip=self.proxy_ip, proxy_port=self.proxy_port)
else:
# 使用pip install pytdx
self.api.connect(ip=self.best_ip['ip'], port=self.best_ip['port'])
self.write_log(f'创建tdx连接, : {self.best_ip}') self.write_log(f'创建tdx连接, : {self.best_ip}')
self.connection_status = True self.connection_status = True

18
vnpy/data/tdx/test_tdx_future.py
View File

@ -14,7 +14,8 @@ from vnpy.data.tdx.tdx_future_data import *
t1 = FakeStrategy() t1 = FakeStrategy()
t2 = FakeStrategy() t2 = FakeStrategy()
# 创建API对象 # 创建API对象
api_01 = TdxFutureData(t1) #api_01 = TdxFutureData(strategy=t1, proxy_ip='localhost', proxy_port=1080)
api_01 = TdxFutureData(strategy=t1)
# 获取所有市场信息 # 获取所有市场信息
markets = api_01.get_markets() markets = api_01.get_markets()
@ -25,8 +26,8 @@ print(u'{}'.format(str_markets))
api_01.qry_instrument() api_01.qry_instrument()
# 获取某个合约得最新价 # 获取某个合约得最新价
price = api_01.get_price('rb2005') #price = api_01.get_price('rb2010')
print('price={}'.format(price)) #print('price={}'.format(price))
# 获取主力合约 # 获取主力合约
@ -63,6 +64,7 @@ corr_rate = round(abs(corr.iloc[0, 1]) * 100, 2)
# api_01.get_bars('IF99', period='1min', callback=t1.display_bar, bar_freq=1) # api_01.get_bars('IF99', period='1min', callback=t1.display_bar, bar_freq=1)
# 获取bar，只返回 list[dict] # 获取bar，只返回 list[dict]
"""
result, bars = api_01.get_bars('IF99', period='1min', return_bar=False) result, bars = api_01.get_bars('IF99', period='1min', return_bar=False)
if result: if result:
print('前十根bar') print('前十根bar')
@ -71,15 +73,15 @@ if result:
print('后十根bar') print('后十根bar')
for bar in bars[-10:]: for bar in bars[-10:]:
print(bar) print(bar)
"""
# result,datas = api_01.get_transaction_data(symbol='ni1905') # result,datas = api_01.get_transaction_data(symbol='ni1905')
# api_02 = TdxFutureData(t2) # api_02 = TdxFutureData(t2)
# api_02.get_bars('IF99', period='1min', callback=t1.display_bar) # api_02.get_bars('IF99', period='1min', callback=t1.display_bar)
# 获取当前交易日分时数据 # 获取当前交易日分时数据
# ret,result = api_01.get_transaction_data('RB99') ret,result = api_01.get_transaction_data('NI99')
# for r in result[0:10] + result[-10:]: for r in result[0:10] + result[-10:]:
# print(r) print(r)
# 获取历史分时数据 # 获取历史分时数据
# ret, result = api_01.get_history_transaction_data('RB99', '20190109') # ret, result = api_01.get_history_transaction_data('RB99', '20190109')
@ -87,4 +89,4 @@ if result:
# print(r) # print(r)
# 更新本地合约缓存信息 # 更新本地合约缓存信息
api_01.update_mi_contracts() #api_01.update_mi_contracts()

9
vnpy/data/tdx/test_tdx_stock.py
View File

@ -17,8 +17,11 @@ import json
t1 = FakeStrategy() t1 = FakeStrategy()
t2 = FakeStrategy() t2 = FakeStrategy()
# 创建API对象
api_01 = TdxStockData(t1) # 创建API对象(使用本地socket5代理）
api_01 = TdxStockData(strategy=t1, proxy_ip='localhost', proxy_port=1080)
# 不使用代理
#api_01 = TdxStockData(strategy=t1)
# 获取市场下股票 # 获取市场下股票
for market_id in range(2): for market_id in range(2):
@ -48,6 +51,6 @@ for market_id in range(2):
# print(r) # print(r)
# 获取历史分时数据 # 获取历史分时数据
ret, result = api_01.get_history_transaction_data('600410', '20190925') ret, result = api_01.get_history_transaction_data('110031', '20200504')
for r in result[0:10] + result[-10:]: for r in result[0:10] + result[-10:]:
print(r) print(r)

BIN
vnpy/gateway/rohon/libLinuxDataCollect.so Normal file
View File

Binary file not shown.

BIN
vnpy/gateway/rohon/libthostmduserapi_se.so Normal file
View File

Binary file not shown.

BIN
vnpy/gateway/rohon/libthosttraderapi_se.so Normal file
View File

Binary file not shown.

1186
vnpy/gateway/rohon/rohon_gateway.py
View File

File diff suppressed because it is too large Load Diff

30
vnpy/gateway/xtp/xtp_gateway.py
View File

@ -1,5 +1,6 @@
from typing import Any, Dict, List from typing import Any, Dict, List
from datetime import datetime from datetime import datetime
from functools import lru_cache
from vnpy.api.xtp import MdApi, TdApi from vnpy.api.xtp import MdApi, TdApi
from vnpy.event import EventEngine from vnpy.event import EventEngine
@ -127,6 +128,9 @@ symbol_name_map: Dict[str, str] = {}
# 代码 <=> 交易所 # 代码 <=> 交易所
symbol_exchange_map: Dict[str, Exchange] = {} symbol_exchange_map: Dict[str, Exchange] = {}
@lru_cache()
def get_vt_symbol_name(vt_symbol):
return symbol_name_map.get(vt_symbol, vt_symbol.split('.')[0])
class XtpGateway(BaseGateway): class XtpGateway(BaseGateway):
@ -238,6 +242,7 @@ class XtpMdApi(MdApi):
self.connect_status: bool = False self.connect_status: bool = False
self.login_status: bool = False self.login_status: bool = False
def onDisconnected(self, reason: int) -> None: def onDisconnected(self, reason: int) -> None:
"""""" """"""
self.connect_status = False self.connect_status = False
@ -298,7 +303,7 @@ class XtpMdApi(MdApi):
tick.bid_volume_1, tick.bid_volume_2, tick.bid_volume_3, tick.bid_volume_4, tick.bid_volume_5 = data["bid_qty"][0:5] tick.bid_volume_1, tick.bid_volume_2, tick.bid_volume_3, tick.bid_volume_4, tick.bid_volume_5 = data["bid_qty"][0:5]
tick.ask_volume_1, tick.ask_volume_2, tick.ask_volume_3, tick.ask_volume_4, tick.ask_volume_5 = data["ask_qty"][0:5] tick.ask_volume_1, tick.ask_volume_2, tick.ask_volume_3, tick.ask_volume_4, tick.ask_volume_5 = data["ask_qty"][0:5]
tick.name = symbol_name_map.get(tick.vt_symbol, tick.symbol) tick.name = get_vt_symbol_name(tick.vt_symbol)
self.gateway.prices.update({tick.vt_symbol: tick.last_price}) self.gateway.prices.update({tick.vt_symbol: tick.last_price})
self.gateway.on_tick(tick) self.gateway.on_tick(tick)
@ -540,6 +545,7 @@ class XtpTdApi(TdApi):
insert_time = str(data["insert_time"]) insert_time = str(data["insert_time"])
dt = datetime.strptime(insert_time, '%Y%m%d%H%M%S%f') dt = datetime.strptime(insert_time, '%Y%m%d%H%M%S%f')
order = OrderData( order = OrderData(
accountid=self.userid,
symbol=symbol, symbol=symbol,
exchange=MARKET_XTP2VT[data["market"]], exchange=MARKET_XTP2VT[data["market"]],
orderid=str(data["order_xtp_id"]), orderid=str(data["order_xtp_id"]),
@ -571,6 +577,7 @@ class XtpTdApi(TdApi):
dt = datetime.strptime(trade_time,'%Y%m%d%H%M%S%f') dt = datetime.strptime(trade_time,'%Y%m%d%H%M%S%f')
trade = TradeData( trade = TradeData(
accountid=self.userid,
symbol=symbol, symbol=symbol,
exchange=MARKET_XTP2VT[data["market"]], exchange=MARKET_XTP2VT[data["market"]],
orderid=str(data["order_xtp_id"]), orderid=str(data["order_xtp_id"]),
@ -615,19 +622,23 @@ class XtpTdApi(TdApi):
if data["market"] == 0: if data["market"] == 0:
return return
vt_symbol = '{}.{}'.format(data["ticker"], MARKET_XTP2VT[data["market"]].value)
position = PositionData( position = PositionData(
accountid=self.userid,
symbol=data["ticker"], symbol=data["ticker"],
exchange=MARKET_XTP2VT[data["market"]], exchange=MARKET_XTP2VT[data["market"]],
name=data["ticker_name"],
direction=POSITION_DIRECTION_XTP2VT[data["position_direction"]], direction=POSITION_DIRECTION_XTP2VT[data["position_direction"]],
volume=data["total_qty"], volume=data["total_qty"],
frozen=data["locked_position"], frozen=data["locked_position"],
price=data["avg_price"], price=data["avg_price"],
pnl=data["unrealized_pnl"], pnl=data["unrealized_pnl"],
yd_volume=data["yesterday_position"], yd_volume=data["yesterday_position"],
gateway_name=self.gateway_name gateway_name=self.gateway_name,
cur_price=self.gateway.prices.get(vt_symbol,0)
) )
vt_symbol = position.vt_symbol if position.volume > 0 and position.cur_price > 0:
position.pnl = round(position.volume * (position.cur_price - position.price),2)
self.gateway.on_position(position) self.gateway.on_position(position)
# 如果持仓>0 获取持仓对应的当前最新价 # 如果持仓>0 获取持仓对应的当前最新价
@ -684,7 +695,8 @@ class XtpTdApi(TdApi):
balance=balance, # 总资产 balance=balance, # 总资产
margin=self.security_asset, # 证券资产 margin=self.security_asset, # 证券资产
frozen=data["withholding_amount"], frozen=data["withholding_amount"],
gateway_name=self.gateway_name gateway_name=self.gateway_name,
trading_day=datetime.now().strftime('%Y-%m-%d')
) )
# AccountData缺省的available 计算方法有误，这里直接取可用资金 # AccountData缺省的available 计算方法有误，这里直接取可用资金
account.available = cash_asset account.available = cash_asset
@ -745,10 +757,12 @@ class XtpTdApi(TdApi):
position = self.short_positions.get(symbol, None) position = self.short_positions.get(symbol, None)
if not position: if not position:
position = PositionData( position = PositionData(
accountid=self.userid,
symbol=symbol, symbol=symbol,
exchange=exchange, exchange=exchange,
direction=Direction.SHORT, direction=Direction.SHORT,
gateway_name=self.gateway_name gateway_name=self.gateway_name,
cur_price=self.gateway.prices.get(f'{symbol}.{exchange.value}',0.0)
) )
self.short_positions[symbol] = position self.short_positions[symbol] = position
@ -855,6 +869,10 @@ class XtpTdApi(TdApi):
orderid = self.insertOrder(xtp_req, self.session_id) orderid = self.insertOrder(xtp_req, self.session_id)
order = req.create_order_data(str(orderid), self.gateway_name) order = req.create_order_data(str(orderid), self.gateway_name)
order.accountid = self.userid
if order.datetime is None:
order.datetime = datetime.now()
order.time = order.datetime.strftime('%H:%M:%S.%f')
self.gateway.on_order(order) self.gateway.on_order(order)
return order.vt_orderid return order.vt_orderid

3
vnpy/trader/object.py
View File

@ -219,6 +219,7 @@ class PositionData(BaseData):
exchange: Exchange exchange: Exchange
direction: Direction direction: Direction
accountid: str = "" # 账号id accountid: str = "" # 账号id
name: str = ""
volume: float = 0 volume: float = 0
frozen: float = 0 frozen: float = 0
price: float = 0 price: float = 0
@ -234,6 +235,8 @@ class PositionData(BaseData):
self.vt_symbol = f"{self.symbol}.{self.exchange.value}" self.vt_symbol = f"{self.symbol}.{self.exchange.value}"
self.vt_positionid = f"{self.gateway_name}.{self.vt_symbol}.{self.direction.value}" self.vt_positionid = f"{self.gateway_name}.{self.vt_symbol}.{self.direction.value}"
self.vt_accountid = f"{self.gateway_name}.{self.accountid}" self.vt_accountid = f"{self.gateway_name}.{self.accountid}"
if self.name == "":
self.name = self.vt_symbol
@dataclass @dataclass
class AccountData(BaseData): class AccountData(BaseData):

1
vnpy/trader/ui/widget.py
View File

@ -460,6 +460,7 @@ class PositionMonitor(BaseMonitor):
sorting = True sorting = True
headers = { headers = {
"name": {"display": "名称", "cell": BaseCell, "update": False},
"symbol": {"display": "代码", "cell": BaseCell, "update": False}, "symbol": {"display": "代码", "cell": BaseCell, "update": False},
"exchange": {"display": "交易所", "cell": EnumCell, "update": False}, "exchange": {"display": "交易所", "cell": EnumCell, "update": False},
"direction": {"display": "方向", "cell": DirectionCell, "update": False}, "direction": {"display": "方向", "cell": DirectionCell, "update": False},