python基于miniQmt打板策略增加止盈止损机制除了2%止盈外增加5%的止损线仓位管理根据市场情况动态调整仓位风险控制单日交易次数限制单只股票仓位限制数据缓存缓存常用的市场数据减少API调用异步处理使用asyncio提高并发处理能力日志系统完善日志记录便于问题排查需要注意的是打板策略风险较高建议先在模拟环境充分测试后再实盘运行。回测数据仅供参考实际交易需考虑滑点、交易成本等因素。策略的成功概率依赖大盘或者板块周期后期应加上这个的影响。#encoding:gbkimport timeimport asyncioclass TradingStrategy:打板交易策略类def __init__(self, path: str, account_id: str, max_cash: float 20000):self.path pathself.account_id account_idself.max_cash max_cashself.xt_trader Noneself.account Noneself.orders {} # 订单管理self.position {} # 持仓管理self.bought_stocks set() # 已买入股票def init_trader(self) - bool:初始化交易接口try:session_id int(time.time())self.xt_trader XtQuantTrader(self.path, session_id)self.account StockAccount(self.account_id, STOCK)self.xt_trader.start()if self.xt_trader.connect() ! 0:raise Exception(交易接口连接失败)print(交易接口连接成功)return Trueexcept Exception as e:print(f初始化交易接口失败: {e})return Falsedef should_buy(self, stock_code: str, stock_data: dict) - bool:优化后的买入条件判断try:current_price stock_data.get(lastPrice, 0)up_stop_price stock_data.get(UpStopPrice, 0)if current_price 0 or up_stop_price 0:return False# 1. 价格接近涨停99.5%以上if current_price up_stop_price * 0.995:return False# 2. 封单金额检查至少3000万bid_volume stock_data.get(bidVol, [0])[0]if bid_volume * up_stop_price * 100 30000000:return False# 3. 流通市值和时间检查float_volume stock_data.get(floatVolume, 0)float_market_value float_volume * current_priceif not self._check_time_by_market_value(float_market_value):return False# 4. 涨幅限制避免追高pre_close stock_data.get(PreClose, 0)if pre_close 0:increase_rate (current_price - pre_close) / pre_closeif increase_rate 0.5: # 涨幅超过50%不追return Falsereturn Trueexcept Exception as e:print(f买入条件判断异常 {stock_code}: {e})return Falsedef _check_time_by_market_value(self, market_value: float) - bool:根据市值检查时间条件market_value_yi market_value / 100000000current_time datetime.now().time()if 30 market_value_yi 100:return current_time datetime.strptime(13:31:00, %H:%M:%S).time()elif 100 market_value_yi 500:return current_time datetime.strptime(14:01:00, %H:%M:%S).time()elif market_value_yi 500:return current_time datetime.strptime(14:51:00, %H:%M:%S).time()return Falsedef monitor_and_trade(self, stock_list: List[str]):优化后的监控交易主循环last_prices {stock: 0 for stock in stock_list}stop_times {stock: 0 for stock in stock_list}while True:try:start_time time.perf_counter()for stock in stock_list:# 跳过已买入的if stock in self.bought_stocks:continuestock_data self._get_stock_data(stock)if not stock_data:continue# 更新价格记录current_price stock_data[lastPrice]last_price last_prices.get(stock, 0)# 买入条件检查if (last_price stock_data[UpStopPrice] andcurrent_price stock_data[UpStopPrice] andself.should_buy(stock, stock_data)):self._execute_buy(stock, stock_data)# 卖出条件检查self._check_sell_conditions(stock, stock_data)# 性能控制elapsed time.perf_counter() - start_timeif elapsed 0.1:time.sleep(0.1 - elapsed)except Exception as e:print(f监控交易异常: {e})time.sleep(1)def _check_sell_conditions(self, stock: str, stock_data: dict):卖出条件检查if stock not in self.position:returnposition_info self.position[stock]current_price stock_data[lastPrice]current_time datetime.now().time()# 条件1: 获利2%卖出buy_price position_info[buy_price]profit_rate (current_price - buy_price) / buy_priceif profit_rate 0.02:self._execute_sell(stock, position_info[volume], current_price)return# 条件2: 10:15后在分时均线上卖出time_1015 datetime.strptime(10:15:00, %H:%M:%S).time()if current_time time_1015:# 获取分时均线价格avg_price self._get_average_price(stock)if avg_price and current_price avg_price:self._execute_sell(stock, position_info[volume], current_price)
python基于miniQmt打板策略增加止盈止损机制除了2%止盈外增加5%的止损线仓位管理根据市场情况动态调整仓位风险控制单日交易次数限制单只股票仓位限制数据缓存缓存常用的市场数据减少API调用异步处理使用asyncio提高并发处理能力日志系统完善日志记录便于问题排查需要注意的是打板策略风险较高建议先在模拟环境充分测试后再实盘运行。回测数据仅供参考实际交易需考虑滑点、交易成本等因素。策略的成功概率依赖大盘或者板块周期后期应加上这个的影响。#encoding:gbkimport timeimport asyncioclass TradingStrategy:打板交易策略类def __init__(self, path: str, account_id: str, max_cash: float 20000):self.path pathself.account_id account_idself.max_cash max_cashself.xt_trader Noneself.account Noneself.orders {} # 订单管理self.position {} # 持仓管理self.bought_stocks set() # 已买入股票def init_trader(self) - bool:初始化交易接口try:session_id int(time.time())self.xt_trader XtQuantTrader(self.path, session_id)self.account StockAccount(self.account_id, STOCK)self.xt_trader.start()if self.xt_trader.connect() ! 0:raise Exception(交易接口连接失败)print(交易接口连接成功)return Trueexcept Exception as e:print(f初始化交易接口失败: {e})return Falsedef should_buy(self, stock_code: str, stock_data: dict) - bool:优化后的买入条件判断try:current_price stock_data.get(lastPrice, 0)up_stop_price stock_data.get(UpStopPrice, 0)if current_price 0 or up_stop_price 0:return False# 1. 价格接近涨停99.5%以上if current_price up_stop_price * 0.995:return False# 2. 封单金额检查至少3000万bid_volume stock_data.get(bidVol, [0])[0]if bid_volume * up_stop_price * 100 30000000:return False# 3. 流通市值和时间检查float_volume stock_data.get(floatVolume, 0)float_market_value float_volume * current_priceif not self._check_time_by_market_value(float_market_value):return False# 4. 涨幅限制避免追高pre_close stock_data.get(PreClose, 0)if pre_close 0:increase_rate (current_price - pre_close) / pre_closeif increase_rate 0.5: # 涨幅超过50%不追return Falsereturn Trueexcept Exception as e:print(f买入条件判断异常 {stock_code}: {e})return Falsedef _check_time_by_market_value(self, market_value: float) - bool:根据市值检查时间条件market_value_yi market_value / 100000000current_time datetime.now().time()if 30 market_value_yi 100:return current_time datetime.strptime(13:31:00, %H:%M:%S).time()elif 100 market_value_yi 500:return current_time datetime.strptime(14:01:00, %H:%M:%S).time()elif market_value_yi 500:return current_time datetime.strptime(14:51:00, %H:%M:%S).time()return Falsedef monitor_and_trade(self, stock_list: List[str]):优化后的监控交易主循环last_prices {stock: 0 for stock in stock_list}stop_times {stock: 0 for stock in stock_list}while True:try:start_time time.perf_counter()for stock in stock_list:# 跳过已买入的if stock in self.bought_stocks:continuestock_data self._get_stock_data(stock)if not stock_data:continue# 更新价格记录current_price stock_data[lastPrice]last_price last_prices.get(stock, 0)# 买入条件检查if (last_price stock_data[UpStopPrice] andcurrent_price stock_data[UpStopPrice] andself.should_buy(stock, stock_data)):self._execute_buy(stock, stock_data)# 卖出条件检查self._check_sell_conditions(stock, stock_data)# 性能控制elapsed time.perf_counter() - start_timeif elapsed 0.1:time.sleep(0.1 - elapsed)except Exception as e:print(f监控交易异常: {e})time.sleep(1)def _check_sell_conditions(self, stock: str, stock_data: dict):卖出条件检查if stock not in self.position:returnposition_info self.position[stock]current_price stock_data[lastPrice]current_time datetime.now().time()# 条件1: 获利2%卖出buy_price position_info[buy_price]profit_rate (current_price - buy_price) / buy_priceif profit_rate 0.02:self._execute_sell(stock, position_info[volume], current_price)return# 条件2: 10:15后在分时均线上卖出time_1015 datetime.strptime(10:15:00, %H:%M:%S).time()if current_time time_1015:# 获取分时均线价格avg_price self._get_average_price(stock)if avg_price and current_price avg_price:self._execute_sell(stock, position_info[volume], current_price)
