# coding:utf-8 from xtquant.xttrader import XtQuantTrader import time, sys from peewee import ModelSelect import xtquant.xtconstant as xtconstant sys.stdout.reconfigure(encoding='utf-8') # 设置标准输出编码为UTF-8 # type: ignore import core.strategy_db as strategy_db import sfgrid_constants from core.sfgrid_trade_controller import StockTradeController from core.util import getInstrumentName, getStockPosition from xtquant.xttrader import XtQuantTrader from xtquant.xttype import StockAccount, XtAsset, XtOrder, XtOrderResponse, XtPosition, XtTrade from xtquant import xtdata from xtquant.xttrader import XtQuantTraderCallback import datetime # 量化核心控制对象 class SFGridController(XtQuantTraderCallback): def __init__(self, account_no: str, miniQmtPath: str): super().__init__() xtdata.enable_hello = False strategy_db.db.connect() strategy_db.db.create_tables([strategy_db.TradeTarget]) print('- [成功]数据库模块初始化') session_id = int(time.time()) self.xt_trader: XtQuantTrader = XtQuantTrader(miniQmtPath, session_id) self.xt_trader.register_callback(self) self.xt_trader.start() connect_result = self.xt_trader.connect() print(f'- [{'成功' if self.xt_trader.connected else '失败'}]市场交易连接{connect_result}--: {miniQmtPath}') if self.xt_trader.connected == False: self.inited: bool = False return else: self.inited = True self.account= StockAccount(account_no, 'STOCK') print(f'- [成功]交易账号对象初始化完成, 账号: {self.account.account_id}') # pyright: ignore[reportAttributeAccessIssue] subscribe_result = self.xt_trader.subscribe(self.account) print(f'- [{'成功' if subscribe_result == 0 else '失败'}:{subscribe_result}]交易状态订阅') if subscribe_result == 0: self.inited = True else: self.inited = False return self.stock_trade_ctrl = {} self.init_instrument_pool(self.xt_trader, self.account) # type: ignore self.seq = None print('- [成功]三疯交易系统初始化完成') self.startMarketData() def startMarketData(self): print('- 启动市场数据订阅') self.seq = xtdata.subscribe_whole_quote(['SH', 'SZ'], callback=self.onDataUpdate) if self.seq == -1: print('- 市场数据订阅失败') else: print(f'- 市场数据订阅成功, 订阅号={self.seq}') def stopMarketData(self): print('- 停止市场数据订阅') if self.seq is not None and self.seq > 0: xtdata.unsubscribe_quote(self.seq) def add_trade_target(self, stock_code: str): try: stock_name = getInstrumentName(stock_code) new_target = strategy_db.TradeTarget.create( stock_name=stock_name, stock_code=stock_code, current_position=0, grid_index=0, last_trade_price=0.0, current_buy_price=0.0, current_buy_order_no='', current_sell_price=0.0, current_sell_order_no='' ) new_target.save() print(f'新增交易标的 {stock_code} {stock_name}, {new_target.id}') # 刷新标的持仓 pos = getStockPosition(stock_code, self.xt_trader, self.account) # type: ignore strategy_db.TradeTarget.update(current_position=pos).where(strategy_db.TradeTarget.stock_code == stock_code).execute() # 更新标的池 self.refresh_targets() # 添加交易控制器 stockTradeController = StockTradeController(new_target, self.xt_trader, self.account, new_target.enabled) # type: ignore self.stock_trade_ctrl[stock_code] = stockTradeController except Exception as e: print(f'新增交易标的失败 {stock_code} {e}') def del_trade_target(self, index:int): target: strategy_db.TradeTarget = self.instrument_pool[index] # self.stock_trade_ctrl. del self.stock_trade_ctrl[target.stock_code] target.delete_instance() self.refresh_targets() def init_instrument_pool(self, xtTrader:XtQuantTrader, account:StockAccount): self.refresh_targets() for temp in self.instrument_pool: tradeTarget:strategy_db.TradeTarget = temp tradeTarget.current_position = getStockPosition(tradeTarget.stock_code, xtTrader, account) # type: ignore result = tradeTarget.save() print(f' |- 同步当前持仓信息 {tradeTarget.stock_code}, {tradeTarget.current_position}, result = {result}') stockTradeController = StockTradeController(tradeTarget, self.xt_trader, self.account, tradeTarget.enabled) # type: ignore self.stock_trade_ctrl[tradeTarget.stock_code] = stockTradeController print(f'- [成功]交易标的信息初始化, 共 {len(self.instrument_pool)} 个标的') def refresh_targets(self): # 更新标的池 self.instrument_pool:ModelSelect = strategy_db.TradeTarget.select() self.print_pool() def print_pool(self): print("- [信息]标的池信息") for i in range(len(self.instrument_pool)): target: strategy_db.TradeTarget = self.instrument_pool[i] status = "新建" if target.status == 0 else "已建初始仓" print(f' [序号-{i}] 股票代码: {target.stock_code}-{target.stock_name} 当前持仓: {getStockPosition(target.stock_code, self.xt_trader, self.account)} 网格索引: {target.grid_index} 基准价格 {sfgrid_constants.grid_price[target.grid_index]} 状态: {status} 启用交易线程: {'自动交易中' if target.enabled else '交易已停止'}') # type: ignore def print_position_info(self): positions:list[XtPosition] = self.xt_trader.query_stock_positions(self.account) if positions: print("\n- 持仓信息") for temp in positions: pos : XtPosition = temp if pos.volume <=0: continue print(f"股票代码: {pos.stock_code}-{getInstrumentName(pos.stock_code)}") print(f"总持仓: {pos.volume}") print(f"可用持仓: {pos.can_use_volume}") print(f"持仓成本: {pos.avg_price}") print("---") else: print("\n当前无持仓") def print_account_info(self): temp = self.xt_trader.query_stock_asset(self.account) asset: XtAsset = temp # type: ignore print(f"=== 账户信息 {self.account.account_id} ===") # type: ignore print(f"可用资金: {asset.cash}") print(f"总资产: {asset.total_asset}") print(f"证券市值: {asset.market_value}") def print_stock_orders(self): orders = self.xt_trader.query_stock_orders(self.account, cancelable_only=True) if orders: print("\n=== 委托信息 ===") for order in orders: print(f"委托编号: {order.order_id}") print(f"股票代码: {order.stock_code} {getInstrumentName(order.stock_code)}") print(f"委托方向: {order.offset_flag} ") print(f"委托价格: {order.price}") print(f"委托数量: {order.order_volume}") print(f"已成交数量: {order.traded_volume}") print(f"委托状态: {order.order_status} ") print("---") else: print("\n当前无委托记录") # 初始化指定标的交易控制器 def start_stock_trade(self, index: int): tradeTarget = self.instrument_pool[index] # check existing thread if tradeTarget.stock_code in self.stock_trade_ctrl: tradeController: StockTradeController = self.stock_trade_ctrl[tradeTarget.stock_code] if tradeController.isEnabled(): print(f"标的交易控制器已存在且正在运行 {tradeTarget.stock_code} {getInstrumentName(tradeTarget.stock_code)}\n") else: print(f"标的交易控制器已存在但未运行，重新启动 {tradeTarget.stock_code} {getInstrumentName(tradeTarget.stock_code)}\n") tradeController.enabledTrading(True) else: stockTradeController = StockTradeController(tradeTarget, self.xt_trader, self.account, tradeTarget.enabled) # type: ignore self.stock_trade_ctrl[tradeTarget.stock_code] = stockTradeController print(f"\t创建标的交易控制器 {tradeTarget.stock_code} {getInstrumentName(tradeTarget.stock_code)}") def pause_stock_trade(self, index: int): tradeTarget = self.instrument_pool[index] if tradeTarget.stock_code in self.stock_trade_ctrl: tradeController: StockTradeController = self.stock_trade_ctrl[tradeTarget.stock_code] if tradeController.isEnabled(): print(f"暂停标的交易 {tradeTarget.stock_code} {getInstrumentName(tradeTarget.stock_code)}\n") tradeController.enabledTrading(False) else: print(f"标的交易已暂停 {tradeTarget.stock_code} {getInstrumentName(tradeTarget.stock_code)}\n") else: print(f"标的交易控制器不存在 {tradeTarget.stock_code} {getInstrumentName(tradeTarget.stock_code)}\n") # ====== 市场回调方法 -- 以下方法由XtQuantData调用 ====== def onDataUpdate(self, data): if sfgrid_constants.max_enabled_targets <= 0: # 全推 for stock_code, tickData in data.items(): lastPrice = tickData['lastPrice'] if lastPrice == 10.0 and stock_code not in self.stock_trade_ctrl: print(f'New trade target = {stock_code} - {getInstrumentName(stock_code)} {tickData['lastPrice']}') self.add_trade_target(stock_code) self.stock_trade_ctrl[stock_code].enabledTrading(True) else: # 指定目标 当前主要使用这种模式 for target in self.instrument_pool: stock_code = target.stock_code # 如果存在对应的StockTradeController，则调用其onDataUpdate方法 if stock_code not in self.stock_trade_ctrl or stock_code not in data: # print(f"股票代码 {stock_code} 未在交易控制器中找到，跳过处理。\n") continue stock_controller: StockTradeController = self.stock_trade_ctrl[stock_code] stock_controller.onDataUpdate(data) # ====== 市场回调方法 -- 以下方法由XtQuantTrader调用 ====== def on_connected(self): """ 连接成功推送 """ print(datetime.datetime.now(), '连接成功回调') def on_disconnected(self): """ 连接断开 :return: """ print(datetime.datetime.now(), '连接断开回调') def on_stock_order(self, order:XtOrder): """ 委托回报推送 :param order: XtOrder对象 :return: """ stockCode = order.stock_code ctrl:StockTradeController = self.stock_trade_ctrl[stockCode] # 如果存在对应的StockTradeController，则调用其onDataUpdate方法 if ctrl is not None and order.strategy_name == ctrl.getName(): ctrl.onOrderTrade(trade=order) # type: ignore else: print(f"委托下单回调 投资备注 orderId: {order.order_sysid} [{order.stock_code}-{order.instrument_name}] volume: {order.order_volume} 订单策略: '{order.strategy_name}'<-->'{ctrl.getName()}'") def test_sim_trade(self, index: int, orderType: int): tradeTarget:strategy_db.TradeTarget = self.instrument_pool[index] ctrl:StockTradeController = self.stock_trade_ctrl[tradeTarget.stock_code] trade: XtTrade = None # type: ignore if orderType == xtconstant.STOCK_BUY: trade = XtTrade( sfgrid_constants.account_no, '300083.SZ', xtconstant.STOCK_BUY, 1, 1, tradeTarget.current_buy_price, sfgrid_constants.grid_volume, 1000, tradeTarget.current_buy_order_no, None, ctrl.getName(), None, None, None, None, None, tradeTarget.stock_name) else: trade = XtTrade(sfgrid_constants.account_no, '300083.SZ', xtconstant.STOCK_SELL, 1, 1, price, sfgrid_constants.grid_volume, 1000, tradeTarget.current_sell_order_no, None, ctrl.getName(), None, None, None, None, None, tradeTarget.stock_name) # type: ignore self.on_stock_trade(trade) def on_stock_trade(self, trade:XtTrade): """ 成交变动推送 :param trade: XtTrade对象 :return: """ stockCode = trade.stock_code ctrl:StockTradeController = self.stock_trade_ctrl[stockCode] # 如果存在对应的StockTradeController，则调用其onDataUpdate方法 if ctrl is not None and trade.strategy_name == ctrl.getName(): ctrl.onOrderTrade(trade) else: print(f"委托回调 投资备注 {trade.strategy_name} 不匹配 {ctrl.getName()}") def on_order_stock_async_response(self, response:XtOrderResponse): stockCode = response.order_remark ctrl:StockTradeController = self.stock_trade_ctrl[stockCode] # 如果存在对应的StockTradeController，则调用其onDataUpdate方法 if ctrl is not None and response.strategy_name == ctrl.getName(): ctrl.onAsyncOrderResponse(response) else: print(f"委托回调 投资备注 {response.strategy_name} 不匹配 {ctrl.getName()}") def on_order_error(self, order_error): """ 委托失败推送 :param order_error:XtOrderError 对象 :return: """ # print("on order_error callback") # print(order_error.order_id, order_error.error_id, order_error.error_msg) print(f"\n委托报错回调 {order_error.order_remark} {order_error.error_msg}") def on_account_status(self, status): """ :param response: XtAccountStatus 对象 :return: """ print(datetime.datetime.now(), sys._getframe().f_code.co_name)