【有问必答专区规则说明】提问必读！

bridges08 · 发表于 2025-4-30 20:58:23

Willows 发表于 2024-3-19 18:35
token完全独立于客户端获取行情，miniqmt本质还是通过客户端获取行情

基于这个情况，token可以用上最新 ...

不是很明白，老师能展开说嘛？谢谢

牛肉汤 · 发表于 2025-5-3 13:33:37

模型交易新建交易策略提示账号没有模型交易功能的权限，不能下单

胖胖 · 发表于 2025-5-6 09:10:41

你们为什么不接入deepseek?,我开通amt有几个月了就是不会用一直没有转入资金,让deepseek写了个策略运行没有交易,不知道哪里问题,希望能接入deepseek,让他把我们的想法变成可执行的代码多好
看看这个为什么不能买入是怎么回事
#encoding:gbk
import pandas as pd
import logging

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# 全局变量
bought_stocks = []  # 已买入的股票列表
max_buy_count = 5 # 当日最大买入股票数量
successful_buys = []  # 成功买入的股票列表
account = "30004991"
buy_amount = 6000  # 每只股票的买入金额

def get_market_data(ContextInfo, fields, stock_code, period, count, subscribe=True):
try:
      return ContextInfo.get_market_data_ex(
         fields=fields,
         stock_code=stock_code,
         period=period,
         count=count,
         subscribe=subscribe
      )
except Exception as e:
      logging.error(f"获取 {stock_code} 的 {fields} 数据时出错: {e}")
      return None

# 初始化函数
def init(ContextInfo):
try:
      # 获取深沪A股和创业板的股票列表
      ContextInfo.stock_list = ContextInfo.get_stock_list_in_sector("沪深A股")
      logging.info(f"股票池大小: {len(ContextInfo.stock_list)}")

      # 订阅股票的实时行情数据
      for stock in ContextInfo.stock_list:
         ContextInfo.subscribe_quote(stock, period="tick")
except Exception as e:
      logging.error(f"初始化时出错: {e}")

# 每个时间步的处理函数
def handlebar(ContextInfo):
try:
      yesterday_close = {}
      yesterday_high = {}
      today_open = {}
      trading_stocks = []  # 用于存储正在交易的股票

      for stock in ContextInfo.stock_list:
         # 获取昨日收盘价和最高价
         data = get_market_data(ContextInfo, ["close", "high"], [stock], "1d", 2)
         if data and stock in data and len(data[stock]) >= 2:
            yesterday_close[stock] = data[stock]["close"].iloc[-2]
            yesterday_high[stock] = data[stock]["high"].iloc[-2]
         else:
            logging.warning(f"未获取到 {stock} 的昨日收盘价或最高价数据")

         # 获取今日开盘价
         data = get_market_data(ContextInfo, ["open"], [stock], "1d", 1)
         if data and stock in data and len(data[stock]) >= 1:
            today_open[stock] = data[stock]["open"].values[0]
         else:
            logging.warning(f"未获取到 {stock} 的今日开盘价数据")

      # 检查股票是否开盘
      for stock in ContextInfo.stock_list:
         try:
            # 获取实时行情数据
            tick_data = ContextInfo.get_full_tick([stock])
            if stock in tick_data and tick_data[stock]["lastPrice"] > 0:
                  trading_stocks.append(stock)
            else:
                  logging.info(f"{stock} 未开盘或无交易数据")
         except Exception as e:
            logging.error(f"检查 {stock} 是否开盘时出错: {e}")

      # 筛选今日开盘价大于昨日最高价且正在交易的股票
      gap_up_stocks = []
      for stock in trading_stocks:
         if stock in yesterday_high and stock in today_open:
            if today_open[stock] > yesterday_high[stock]:
                  gap_up_stocks.append(stock)

      # 在跳空高开的股票中，筛选涨幅大于等于3%的股票
      high_gap_up_stocks = []
      for stock in gap_up_stocks:
         if stock in yesterday_close:
            yesterday_close_price = yesterday_close[stock]
            today_open_price = today_open[stock]
            gap_percentage = (today_open_price - yesterday_close_price) / yesterday_close_price * 100
            if gap_percentage >= 3:
                  high_gap_up_stocks.append(stock)
                  logging.info(f"{stock} 跳空高开且涨幅大于等于3%，昨日收盘价: {yesterday_close_price}, 今日开盘价: {today_open_price}, 跳空幅度: {gap_percentage:.2f}%")

      # 统计选出的股票总数
      high_gap_up_count = len(high_gap_up_stocks)
      logging.info(f"跳空高开且涨幅大于等于3%的股票总数: {high_gap_up_count}")
      logging.info(f"跳空高开且涨幅大于等于3%的股票列表: {high_gap_up_stocks}")

      # 对筛选出的股票进行实时监控
      for stock in high_gap_up_stocks:
         ContextInfo.subscribe_quote(stock, period="tick", callback=lambda data, s=stock: monitor_stock(data, s))
except Exception as e:
      logging.error(f"处理时间步时出错: {e}")

# 实时监控股票是否涨停
def monitor_stock(data, stock):
global bought_stocks, successful_buys
try:
      if len(bought_stocks) >= max_buy_count:
         logging.info(f"已达到当日最大买入数量 {max_buy_count}，不再买入新股票")
         return

      last_price = data[stock]["lastPrice"]
      last_close = data[stock]["lastClose"]
      upper_limit = round(last_close * 1.1, 2)  # 涨停价计算

      if last_price >= upper_limit:
         logging.info(f"{stock} 涨停，尝试买入")
         if buy_stock(stock, upper_limit, ContextInfo):
            successful_buys.append(stock)  # 添加到成功买入列表
      elif stock in bought_stocks and last_price < upper_limit:
         logging.info(f"{stock} 涨停打开，等待再次封涨停")
         # 设置一个短期定时任务，监控涨停是否再次封住
         ContextInfo.run_time(lambda C, s=stock, ul=upper_limit: check_re_seal(C, s, ul), "3nSecond")
except Exception as e:
      logging.error(f"监控 {stock} 时出错: {e}")

# 检查股票是否再次封涨停
def check_re_seal(ContextInfo, stock, upper_limit):
if stock not in bought_stocks:
      return

try:
      data = ContextInfo.get_full_tick([stock])
      if data[stock]["lastPrice"] >= upper_limit:
         logging.info(f"{stock} 再次封涨停，尝试买入")
         if buy_stock(stock, upper_limit, ContextInfo):
            successful_buys.append(stock)  # 添加到成功买入列表
except Exception as e:
      logging.error(f"检查 {stock} 再次封涨停时出错: {e}")

# 买入股票
def buy_stock(stock, price, ContextInfo):
global bought_stocks
if len(bought_stocks) >= max_buy_count:
      logging.info(f"已达到当日最大买入数量 {max_buy_count}，不再买入新股票")
      return False

if stock not in bought_stocks:
      bought_stocks.append(stock)
      # 计算买入股数，向下取整到100的倍数
      quantity = int((buy_amount // price) // 100) * 100
      if quantity > 0:
         logging.info(f"买入 {stock}，价格: {price}，数量: {quantity} 股")
         # 下单逻辑（示例）
         # 使用 passorder 下单
         passorder(23, 1101, account, stock, 5, price, quantity, "跳空高开涨停买入", 2, f"买入 {stock}", ContextInfo)
         return True
      else:
         logging.info(f"按 {price} 元的价格，{buy_amount} 元不足以买入 {stock}，无法下单")
return False

# 在每个时间步结束时打印成功买入的股票列表
def handlebar_end(ContextInfo):
logging.info(f"成功买入的股票列表: {successful_buys}")