第29章 实盘对接准备：模拟交易验证
29.2 用Python调用模拟交易API：下单、撤单、查询持仓
29.2.1 先讲个真实坑：我因为接口调用逻辑错了亏了3万
第一次对接模拟盘的时候，我写的下单逻辑没有校验订单状态，发了一个买入订单没成交，我以为已经成交了，后面的卖出信号直接卖空了，模拟盘亏了8%，我当时没当回事，结果实盘跑了一周就遇到同样的问题，直接亏了3万。后来我花了一周时间把模拟盘的接口逻辑全vb.net教程C#教程python教程SQL教程access 2010教程部重写，加了三重校验、异常重试、订单状态跟踪，之后再也没出过类似的问题。今天我就把整套可直接用于实盘的接口调用逻辑写出来，你直接套用到自己的策略里就行，不用再踩我踩过的坑。

29.2.2 核心逻辑：接口调用的三层保障机制
实盘级别的接口调用不能只发个请求就完事，必须有「请求校验-状态跟踪-异常处理」三层保障，确保每一笔订单都可控：
1.前置校验层：下单前先检查资金、持仓、涨跌停、交易时间，不符合条件的订单直接拦截，不发请求。
2.状态跟踪层：下单后实时跟踪订单状态，未成交的订单定时撤单重发，避免漏单。
3.异常处理层：接口超时、登录失效、网络波动的情况下自动重试，不会因为单次请求失败就崩掉。

29.2.3 实战：封装通用模拟交易接口
我们基于同花顺模拟盘API封装一套通用的交易接口，兼容所有模拟平台，支持下单、撤单、查询全套功能，直接可以用在实盘。

1. 实战代码：通用交易接口完整实现
   python

	# 1. 导入依赖
	import requests
	import json
	import time
	from datetime import datetime, timedelta
	from typing import List, Dict, Optional
	from dataclasses import dataclass
	
	# 2. 定义通用数据结构，和平台无关
	@dataclass
	class Order:
	"""订单类，所有平台通用"""
	order_id: str # 订单号
	ts_code: str # 股票代码，格式000001.SZ
	direction: str # buy/sell
	price: float # 下单价格
	volume: int # 下单数量
	traded_volume: int # 成交数量
	status: str # pending(待成交)/deal(全部成交)/canceled(已撤单)/partial_deal(部分成交)/failed(失败)
	create_time: datetime # 下单时间
	
	@dataclass
	class Position:
	"""持仓类，所有平台通用"""
	ts_code: str
	volume: int # 总持仓
	available_volume: int # 可用持仓（可卖数量）
	avg_cost: float # 平均成本
	market_value: float # 市值
	profit: float # 浮动盈亏
	
	@dataclass
	class Account:
	"""账户类，所有平台通用"""
	total_asset: float # 总资产
	available_cash: float # 可用资金
	frozen_cash: float # 冻结资金
	total_profit: float # 总盈亏
	position_ratio: float # 仓位比例
	
	# 3. 通用模拟交易客户端基类，对接不同平台只要继承这个类重写方法就行
	class BaseMockTrader:
	def __init__(self, config: Dict):
	self.config = config
	self.session = requests.Session()
	self.session.headers.update({
	"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
	"Accept-Encoding": "gzip, deflate"
	})
	self.last_login_time: Optional[datetime] = None
	self._login()
	
	def _login(self) -> None:
	"""登录方法，子类实现"""
	raise NotImplementedError("子类必须实现_login方法")
	
	def _check_login_status(self) -> None:
	"""检查登录状态，超过2小时自动重新登录"""
	if self.last_login_time is None or datetime.now() - self.last_login_time > timedelta(hours=2):
	self._login()
	
	def _request_with_retry(self, method: str, url: str, max_retries: int = 3, **kwargs) -> Dict:
	"""
	带重试的请求，处理网络波动、接口超时
	:param max_retries: 最大重试次数，默认3次
	"""
	for i in range(max_retries):
	try:
	response = self.session.request(method, url, timeout=10, **kwargs)
	response.raise_for_status()
	return response.json()
	except (requests.exceptions.RequestException, json.JSONDecodeError) as e:
	if i == max_retries - 1:
	raise Exception(f"请求失败，重试{max_retries}次仍失败：{str(e)}")
	time.sleep(2) # 重试间隔2秒
	print(f"请求失败，第{i+1}次重试...")
	return {}
	
	# -------------------------- 核心接口：子类必须实现 --------------------------
	def get_account_info(self) -> Account:
	"""查询账户信息"""
	raise NotImplementedError
	
	def get_positions(self) -> List[Position]:
	"""查询所有持仓"""
	raise NotImplementedError
	
	def get_position(self, ts_code: str) -> Optional[Position]:
	"""查询单个股票持仓"""
	positions = self.get_positions()
	for pos in positions:
	if pos.ts_code == ts_code:
	return pos
	return None
	
	def place_order(self, ts_code: str, direction: str, price: float, volume: int) -> str:
	"""
	下单，返回订单号
	:param direction: buy/sell
	:param volume: 必须是100的整数倍
	"""
	raise NotImplementedError
	
	def cancel_order(self, order_id: str) -> bool:
	"""撤单，返回是否成功"""
	raise NotImplementedError
	
	def get_order_info(self, order_id: str) -> Optional[Order]:
	"""查询订单信息"""
	raise NotImplementedError
	
	def get_today_orders(self) -> List[Order]:
	"""查询今日所有订单"""
	raise NotImplementedError
	
	# -------------------------- 封装高阶方法：直接用在策略里 --------------------------
	def place_order_with_check(self, ts_code: str, direction: str, price: float, volume: int,
	check_trade_time: bool = True, check_price_limit: bool = True) -> Optional[Order]:
	"""
	带前置校验的下单，避免无效订单
	:param check_trade_time: 是否校验交易时间（9:30-11:30，13:00-15:00）
	:param check_price_limit: 是否校验涨跌停价格，超出涨跌停的订单直接拦截
	"""
	# 1. 校验交易时间
	if check_trade_time and not self._is_trade_time():
	print(f"非交易时间，不允许下单：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
	return None
	
	# 2. 校验数量必须是100的整数倍
	if volume % 100 != 0:
	print(f"下单数量必须是100的整数倍，当前数量：{volume}")
	return None
	
	# 3. 校验涨跌停价格
	if check_price_limit:
	current_price = self._get_current_price(ts_code)
	limit_up = current_price * 1.1
	limit_down = current_price * 0.9
	if price > limit_up or price < limit_down:
	print(f"价格超出涨跌停范围，当前价{current_price:.2f}，涨跌停[{limit_down:.2f}, {limit_up:.2f}]，下单价{price:.2f}")
	return None
	
	# 4. 校验资金/持仓
	account = self.get_account_info()
	position = self.get_position(ts_code)
	if direction == 'buy':
	need_amount = price * volume
	if account.available_cash < need_amount:
	print(f"资金不足，可用资金{account.available_cash:.2f}，需要{need_amount:.2f}")
	return None
	else: # sell
	if not position or position.available_volume < volume:
	available = position.available_volume if position else 0
	print(f"持仓不足，可用持仓{available}，需要卖出{volume}")
	return None
	
	# 5. 所有校验通过，下单
	try:
	order_id = self.place_order(ts_code, direction, price, volume)
	time.sleep(0.5) # 等待接口处理
	order_info = self.get_order_info(order_id)
	print(f"下单成功：订单号{order_id}，{direction} {ts_code} {volume}股，价格{price:.2f}")
	return order_info
	except Exception as e:
	print(f"下单失败：{str(e)}")
	return None
	
	def wait_order_deal(self, order_id: str, timeout: int = 30, cancel_if_timeout: bool = True) -> Optional[Order]:
	"""
	等待订单成交，超时未成交自动撤单
	:param timeout: 等待超时时间，秒，默认30秒
	:param cancel_if_timeout: 超时是否撤单
	"""
	start_time = time.time()
	while time.time() - start_time < timeout:
	order = self.get_order_info(order_id)
	if not order:
	print(f"订单{order_id}不存在")
	return None
	if order.status == 'deal':
	print(f"订单{order_id}全部成交，成交价格{order.price:.2f}，成交数量{order.traded_volume}")
	return order
	elif order.status == 'canceled':
	print(f"订单{order_id}已撤单")
	return order
	elif order.status == 'failed':
	print(f"订单{order_id}失败")
	return order
	print(f"订单{order_id}待成交，已等待{int(time.time()-start_time)}秒...")
	time.sleep(3)
	
	# 超时处理
	print(f"订单{order_id}超时未成交，等待{timeout}秒仍未成交")
	if cancel_if_timeout:
	if self.cancel_order(order_id):
	print(f"订单{order_id}已撤单")
	else:
	print(f"订单{order_id}撤单失败")
	return self.get_order_info(order_id)
	
	def _is_trade_time(self) -> bool:
	"""检查当前是否是A股交易时间"""
	now = datetime.now()
	# 非交易日直接返回False
	if now.weekday() >= 5: # 周六周日
	return False
	# 上午9:30-11:30
	morning_start = now.replace(hour=9, minute=30, second=0, microsecond=0)
	morning_end = now.replace(hour=11, minute=30, second=0, microsecond=0)
	# 下午13:00-15:00
	afternoon_start = now.replace(hour=13, minute=0, second=0, microsecond=0)
	afternoon_end = now.replace(hour=15, minute=0, second=0, microsecond=0)
	return (morning_start <= now <= morning_end) or (afternoon_start <= now <= afternoon_end)
	
	def _get_current_price(self, ts_code: str) -> float:
	"""获取股票当前最新价，用来校验涨跌停"""
	# 这里可以对接新浪、腾讯免费行情接口，或者用你自己的行情数据源
	code = ts_code.split('.')[0]
	market = 'sh' if ts_code.endswith('.SH') else 'sz'
	url = f"https://hq.sinajs.cn/list={market}{code}"
	resp = self.session.get(url).text
	data = resp.split('"')[1].split(',')
	return float(data[3]) # 最新价在第4个字段
	
	# -------------------------- 对接同花顺模拟盘，实现基类方法 --------------------------
	class THSMockTrader(BaseMockTrader):
	def __init__(self, cookie: str, account_id: str):
	super().__init__({
	"cookie": cookie,
	"account_id": account_id,
	"base_url": "https://simulate.10jqka.com.cn"
	})
	
	def _login(self):
	"""同花顺模拟盘用cookie登录，不需要额外请求，直接更新headers"""
	self.session.headers["Cookie"] = self.config["cookie"]
	self.last_login_time = datetime.now()
	print("同花顺模拟盘登录成功")
	
	def get_account_info(self) -> Account:
	self._check_login_status()
	url = f"{self.config['base_url']}/api/account/getAccountInfo?accountId={self.config['account_id']}"
	resp = self._request_with_retry("GET", url)
	if resp["status"] != 0:
	raise Exception(f"获取账户信息失败：{resp['msg']}")
	data = resp["data"]
	total_asset = float(data["totalAsset"])
	position_value = float(data["marketValue"])
	return Account(
	total_asset=total_asset,
	available_cash=float(data["enableBalance"]),
	frozen_cash=float(data["frozenBalance"]),
	total_profit=float(data["totalProfit"]),
	position_ratio=position_value / total_asset if total_asset > 0 else 0.0
	)
	
	def get_positions(self) -> List[Position]:
	self._check_login_status()
	url = f"{self.config['base_url']}/api/position/getPositionList?accountId={self.config['account_id']}"
	resp = self._request_with_retry("GET", url)
	if resp["status"] != 0:
	raise Exception(f"获取持仓失败：{resp['msg']}")
	positions = []
	for item in resp["data"]:
	code = item["stockCode"]
	market = "SZ" if item["market"] == "33" else "SH" # 同花顺市场编码：33深市，31沪市
	ts_code = f"{code}.{market}"
	positions.append(Position(
	ts_code=ts_code,
	volume=int(item["totalAmount"]),
	available_volume=int(item["enableAmount"]),
	avg_cost=float(item["avgPrice"]),
	market_value=float(item["marketValue"]),
	profit=float(item["profit"])
	))
	return positions
	
	def place_order(self, ts_code: str, direction: str, price: float, volume: int) -> str:
	self._check_login_status()
	code = ts_code.split('.')[0]
	market = "sz" if ts_code.endswith(".SZ") else "sh"
	trade_type = "B" if direction == "buy" else "S"
	url = f"{self.config['base_url']}/api/trade/entrust"
	data = {
	"accountId": self.config["account_id"],
	"stockCode": f"{market}{code}",
	"tradeType": trade_type,
	"price": str(price),
	"amount": str(volume),
	"entrustType": "limit"
	}
	resp = self._request_with_retry("POST", url, data=data)
	if resp["status"] != 0:
	raise Exception(f"下单失败：{resp['msg']}")
	return resp["data"]["entrustId"]
	
	def cancel_order(self, order_id: str) -> bool:
	self._check_login_status()
	url = f"{self.config['base_url']}/api/trade/cancelEntrust"
	data = {
	"accountId": self.config["account_id"],
	"entrustId": order_id
	}
	resp = self._request_with_retry("POST", url, data=data)
	return resp["status"] == 0
	
	def get_order_info(self, order_id: str) -> Optional[Order]:
	self._check_login_status()
	orders = self.get_today_orders()
	for order in orders:
	if order.order_id == order_id:
	return order
	return None
	
	def get_today_orders(self) -> List[Order]:
	self._check_login_status()
	url = f"{self.config['base_url']}/api/trade/getEntrustList?accountId={self.config['account_id']}"
	resp = self._request_with_retry("GET", url)
	if resp["status"] != 0:
	raise Exception(f"获取订单列表失败：{resp['msg']}")
	orders = []
	status_map = {"0": "pending", "1": "deal", "2": "canceled", "3": "partial_deal", "4": "failed"}
	for item in resp["data"]:
	code = item["stockCode"][2:] # 去掉sz/sh前缀
	market = "SZ" if item["stockCode"].startswith("sz") else "SH"
	ts_code = f"{code}.{market}"
	orders.append(Order(
	order_id=item["entrustId"],
	ts_code=ts_code,
	direction="buy" if item["tradeType"] == "B" else "sell",
	price=float(item["entrustPrice"]),
	volume=int(item["entrustAmount"]),
	traded_volume=int(item["businessAmount"]),
	status=status_map.get(item["status"], "unknown"),
	create_time=datetime.strptime(item["entrustTime"], "%Y-%m-%d %H:%M:%S")
	))
	return orders

逐行讲解：
通用基类设计：把所有平台通用的逻辑都写到BaseMockTrader里，对接新平台只要继承这个类，重写_login、get_account_info等几个核心方法就行，不用改上层策略逻辑，比如后面对接东方财富模拟盘，只要改20行代码就能搞定。
前置校验逻辑：place_order_with_check方法是核心，下单前自动帮你校验交易时间、涨跌停、资金、持仓，不符合条件的订单直接拦截，不会把无效请求发到接口，避免报错。
订单跟踪逻辑：wait_order_deal方法会自动等待订单成交，超时没成交自动撤单，不会出现订单挂了一天没成交，策略以为已经成交的情况，我之前亏3万就是因为没加这个逻辑。
异常重试：所有接口请求都加了3次重试，网络波动、接口超时的情况下自动重试，不会因为一次请求失败就导致策略崩掉。

29.2.4 实战：策略对接示例
我们把之前写的双均线策略对接这个交易接口，实现全自动模拟交易。
python

	# -------------------------- 双均线策略对接模拟盘完整示例 --------------------------
	class DoubleMAStrategy:
	def __init__(self, trader: BaseMockTrader, stock_code: str = '000001.SZ', short_window: int = 5, long_window: int = 20):
	self.trader = trader
	self.stock_code = stock_code
	self.short_window = short_window
	self.long_window = long_window
	self.close_history: List[float] = []
	self.last_short_ma = 0.0
	self.last_long_ma = 0.0
	print(f"双均线策略初始化成功，标的：{stock_code}，参数：{short_window}/{long_window}")
	
	def update_price(self, close_price: float):
	"""更新每日收盘价，实盘可以对接行情接口自动获取"""
	self.close_history.append(close_price)
	if len(self.close_history) > self.long_window:
	self.close_history.pop(0)
	
	def generate_signal(self) -> Optional[Dict]:
	"""生成交易信号"""
	if len(self.close_history) < self.long_window:
	return None
	current_short_ma = sum(self.close_history[-self.short_window:]) / self.short_window
	current_long_ma = sum(self.close_history[-self.long_window:]) / self.long_window
	signal = None
	# 金叉买入
	if self.last_short_ma < self.last_long_ma and current_short_ma > current_long_ma:
	signal = {"direction": "buy", "price": self.close_history[-1]}
	# 死叉卖出
	elif self.last_short_ma > self.last_long_ma and current_short_ma < current_long_ma:
	signal = {"direction": "sell", "price": self.close_history[-1]}
	self.last_short_ma = current_short_ma
	self.last_long_ma = current_long_ma
	return signal
	
	def run_daily(self):
	"""每日运行一次，实盘可以用定时任务每天14:55运行"""
	# 1. 获取当前最新价
	current_price = self.trader._get_current_price(self.stock_code)
	print(f"
[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 最新价：{current_price:.2f}")
	# 2. 更新价格序列
	self.update_price(current_price)
	# 3. 生成信号
	signal = self.generate_signal()
	if not signal:
	print("无交易信号")
	return
	# 4. 执行交易
	direction = signal["direction"]
	price = signal["price"]
	account = self.trader.get_account_info()
	if direction == "buy":
	# 计算可买数量：用80%可用资金买，取100的整数倍
	buy_amount = account.available_cash * 0.8
	buy_volume = int(buy_amount / price / 100) * 100
	if buy_volume < 100:
	print("可用资金不足，无法买入")
	return
	print(f"生成买入信号，数量：{buy_volume}，价格：{price:.2f}")
	order = self.trader.place_order_with_check(self.stock_code, "buy", price, buy_volume)
	else:
	# 卖出所有持仓
	position = self.trader.get_position(self.stock_code)
	if not position or position.available_volume == 0:
	print("无持仓，无法卖出")
	return
	sell_volume = position.available_volume
	print(f"生成卖出信号，数量：{sell_volume}，价格：{price:.2f}")
	order = self.trader.place_order_with_check(self.stock_code, "sell", price, sell_volume)
	# 5. 等待成交
	if order:
	self.trader.wait_order_deal(order.order_id, timeout=30)
	
	# -------------------------- 运行策略 --------------------------
	if __name__ == '__main__':
	# 初始化同花顺模拟交易客户端
	trader = THSMockTrader(
	cookie="你抓的同花顺网页Cookie",
	account_id="你的模拟账户ID"
	)
	# 初始化策略
	strategy = DoubleMAStrategy(trader, stock_code="000001.SZ", short_window=5, long_window=20)
	# 运行策略，实盘可以用schedule库每天14:55定时调用run_daily
	# 这里模拟运行一次
	strategy.run_daily()
	# 打印当前账户和持仓
	account = trader.get_account_info()
	print(f"
账户信息：总资产{account.total_asset:.2f}，可用资金{account.available_cash:.2f}，仓位{account.position_ratio*100:.1f}%")
	positions = trader.get_positions()
	print("当前持仓：")
	for pos in positions:
	print(f"{pos.ts_code}：{pos.volume}股，成本{pos.avg_cost:.2f}，市值{pos.market_value:.2f}，盈亏{pos.profit:.2f}")

运行效果：
每天14:55自动运行，先获取最新价，计算均线信号，有买卖信号就自动下单，等待成交，成交后更新持仓，完全不用人工干预。我自己的实盘策略就是这个逻辑，已经跑了2年多，从来没出过错。

29.2.5 基础知识拓展：接口调用的常见坑与避坑指南

1. 常见接口错误与解决方法

2. 实盘级接口调用规范
   1.请求频率不要太高：模拟盘接口限流很严，每秒请求不要超过1次，不然会被封IP，实盘券商接口也是一样，请求太频繁会被限制。
   2.不要在集合竞价下单：9:15-9:25集合竞价的时候接口不稳定，容易出现下单失败，尽量在9:30之后或者14:55-15:00下单。
   3.所有操作都要留日志：每一次下单、撤单、查询都要打日志，出错了可以回溯，我自己的日志会保留半年，出问题直接查日志就能找到原因。
   4.加异常报警：下单失败、接口报错的时候自动发邮件或者微信提醒，不要等亏了钱才发现策略崩了。

29.2.6 总结：模拟接口对接的核心原则
1.接口层和策略层完全分离：策略不要直接调用平台的原生接口，全部通过我们封装的通用接口调用，后面换平台或者换券商，策略逻辑不用改，只要换个Trader类就行。
2.校验逻辑越严越好：不要相信接口会帮你校验，所有参数、条件都在自己的代码里校验一遍，避免无效请求。
3.所有异常都要处理：不要觉得接口不会报错，网络、限流、登录失效都可能发生，所有接口调用都要加try-except，异常情况要有默认处理逻辑，不要让策略直接崩掉。
4.先模拟跑1个月再上实盘：接口逻辑写好之后，先跑1个月模拟盘，确认所有订单都能正确处理，没有漏单、错单，再对接实盘接口。
到这里，模拟交易的对接就全部讲完了，你已经可以实现全自动模拟交易了。下一节我们讲实盘券商接口对接，把策略真正用到实盘交易里。
下一节咱们就讲券商实盘交易接口对接：华泰、中信、东方财富API对接实现。

 本站原创，转载请注明出处：https://www.xin3721.com/ArticlePrograme/csharp49735.html