【Python技术】量化之miniqmt根据分时均线买入卖出示例

import timeimport mathimport threadingimport pandas as pdfrom datetime import datetime, time as dt_timefrom xtquant import xtdata, xtconstantfrom xtquant.xttype import StockAccountfrom xtquant.xttrader import XtQuantTrader, XtQuantTraderCallbackclass VWAPStrategy:    def __init__(self, stock_code, capital_per_order=10000):        self.stock_code = stock_code        self.capital_per_order = capital_per_order        self.position = 0        self.vwap = 0        self.last_trade_time = None        self.position_entry_price = 0  # 持仓成本价        self.position_entry_time = None  # 持仓时间    def fetch_stock_data(self):        """获取实时数据并计算分时均线(VWAP)"""        tick_data = xtdata.get_full_tick([self.stock_code])        if self.stock_code in tick_data:            stock_tick = tick_data[self.stock_code]            total_volume = stock_tick['volume']  # 总成交量（手）            total_amount = stock_tick['amount']  # 总成交金额（元）            current_price = stock_tick['lastPrice']  # 最新价            # 计算分时均线（VWAP）            self.vwap = total_amount / (total_volume * 100) if total_volume > 0 else current_price            return {                'price': current_price,                'vwap': self.vwap,                'volume': total_volume,                'amount': total_amount,                'time': datetime.now().strftime("%H:%M:%S")            }        return None    def generate_signal(self, data):        """生成交易信号"""        current_price = data['price']        vwap = data['vwap']        deviation = (current_price - vwap) / vwap * 100  # 计算偏离度        # 买入条件1：价格回落未破分时均线 (当前价>VWAP且回落≤0.5%)        buy_condition1 = (current_price > vwap and                          (vwap - current_price) / vwap <= 0.005)        # 买入条件2：分时均线下方偏离超过5%        buy_condition2 = deviation < -5        # 卖出条件1：跌破分时均线止损        sell_condition1 = (self.position > 0 and                           current_price < vwap)        # 卖出条件2：分时价格超过均线5%止盈        sell_condition2 = (self.position > 0 and                           deviation > 5)        


    
# 卖出条件3：超短线持仓超过3天        sell_condition3 = (self.position > 0 and                           self.position_entry_time and                           (datetime.now() - self.position_entry_time).days >= 3)        if buy_condition1:            return 'BUY', f"回落未破均线({deviation:.2f}%)"        elif buy_condition2:            return 'BUY', f"均线下偏离({deviation:.2f}%)"        elif sell_condition1:            return 'SELL', f"跌破均线止损({deviation:.2f}%)"        elif sell_condition2:            return 'SELL', f"偏离止盈({deviation:.2f}%)"        elif sell_condition3:            return 'SELL', "超时持仓(≥3天)"        return 'HOLD', ""    def execute_trade(self, xt_trader, account, signal, reason, current_price):        """执行交易指令"""        # 30秒内禁止重复交易        if self.last_trade_time and (datetime.now() - self.last_trade_time).seconds 30:            return False        # 计算交易数量 (按固定资金计算)        volume = math.floor(self.capital_per_order / current_price / 100) * 100  # 整手交易        if signal == 'BUY' and volume > 0:            # 调用MiniQMT买入接口            order_id = xt_trader.order_stock(                account, self.stock_code,                xtconstant.STOCK_BUY, volume,                xtconstant.LATEST_PRICE, 0,                f"策略买入:{reason}"            )            if order_id:                self.position += volume                self.position_entry_price = current_price                self.position_entry_time = datetime.now()                print(f"[{datetime.now().strftime('%H:%M:%S')}] 买入信号触发: {reason}")                return True        elif signal == 'SELL' and self.position > 0:            sell_volume = min(volume, self.position)            # 调用MiniQMT卖出接口            order_id = xt_trader.order_stock(                account, self.stock_code,                xtconstant.STOCK_SELL, sell_volume,                xtconstant.LATEST_PRICE, 0,                f"策略卖出:{reason}"            )            if order_id:                self.position = max(0, self.position - sell_volume)                if self.position == 0:                    self.position_entry_price = 0                    self.position_entry_time = None                print(f"[{datetime.now().strftime('%H:%M:%S')}] 卖出信号触发: {reason}")


    
                return True        self.last_trade_time = datetime.now()        return Falseclass MyXtQuantTraderCallback(XtQuantTraderCallback):    """交易回调处理类"""    def on_disconnected(self):        print(f"[{datetime.now().strftime('%H:%M:%S')}] 连接断开")    def on_stock_order(self, order):        print(f"[{datetime.now().strftime('%H:%M:%S')}] 委托更新: {order.stock_code} "              f"状态={self._get_order_status(order.order_status)}")    def on_stock_trade(self, trade):        print(f"[{datetime.now().strftime('%H:%M:%S')}] 成交: {trade.stock_code} "              f"{trade.traded_volume}股@{trade.traded_price:.2f}")    def on_order_error(self, order_error):        print(f"[{datetime.now().strftime('%H:%M:%S')}] 委托错误: {order_error.error_msg}")    @staticmethod    def _get_order_status(status):        """转换订单状态码为可读文本"""        status_map = {            48: '未报', 49: '待报', 50: '已报',            51: '已报待撤', 52: '部分待撤', 53: '部撤',            54: '已撤', 55: '部成', 56: '已成', 57: '废单'        }        return status_map.get(status, '未知状态')def is_trading_time():    """判断是否在交易时段"""    now = datetime.now()    return (dt_time(9, 30) <= now.time() <= dt_time(15, 0)) and now.weekday() 5def main():    # ========== 配置参数 ==========    QMT_PATH = r"D:/国金QMT交易端模拟/userdata_mini"  # 替换为你的MiniQMT路径    ACCOUNT_ID = "XXXXXXX"  # 替换为你的资金账号    STOCK_CODE = "XXXXXX.SZ"  # 监控的股票代码    CAPITAL_PER_ORDER = 20000  # 单笔交易金额(元)    # ========== 初始化交易接口 ==========    session_id = int(time.time())  # 唯一会话ID    xt_trader = XtQuantTrader(QMT_PATH, session_id)    account = StockAccount(ACCOUNT_ID, 'STOCK')    # 创建并注册回调    callback = MyXtQuantTraderCallback()    xt_trader.register_callback(callback)    # 启动交易线程    xt_trader.start()    # 建立交易连接    connect_result = xt_trader.connect()    if connect_result != 0:        print(f"交易连接失败，错误代码: {connect_result}


    
")        return    # 订阅账户    subscribe_result = xt_trader.subscribe(account)    if subscribe_result != 0:        print(f"账户订阅失败，错误代码: {subscribe_result}")        return    print("=" * 50)    print(f"启动分时均线策略监控: {STOCK_CODE}")    print(f"交易账号: {ACCOUNT_ID} | 单笔金额: {CAPITAL_PER_ORDER}元")    print("=" * 50)    # ========== 初始化策略 ==========    strategy = VWAPStrategy(STOCK_CODE, CAPITAL_PER_ORDER)    # ========== 主循环 ==========    try:        while True:            if not is_trading_time():                # 非交易时段每30秒检查一次                print(f"\r[{datetime.now().strftime('%H:%M:%S')}] 非交易时段，等待开盘...", end="")                time.sleep(30)                continue            # 获取实时数据            data = strategy.fetch_stock_data()            if not data:                time.sleep(3)                continue            # 生成交易信号            signal, reason = strategy.generate_signal(data)            # 输出监控信息            status_line = (f"[{data['time']}] 价格:{data['price']:.2f} | "                           f"均线:{data['vwap']:.2f} | "                           f"偏离:{((data['price'] - data['vwap']) / data['vwap'] * 100):.2f}% | "                           f"持仓:{strategy.position}股")            if signal != 'HOLD':                print(f"{status_line} → 信号: {signal}({reason})")                strategy.execute_trade(xt_trader, account, signal, reason, data['price'])            else:                print(status_line)            time.sleep(3)  # 3秒刷新    except KeyboardInterrupt:        print("\n策略终止")    finally:        xt_trader.stop()if __name__ == "__main__":    main()