第30章 券商实盘交易接口对接
30.1 券商API对接：华泰Level2、中信建投、国泰君安接口
30.1.1 先讲个踩坑经历：我因为对接错券商接口亏了12万
2021年我第一次对接实盘，找了个第三方破解的券商接口，用了半个月都正常，结果有一天接口突然失效，卖出信号发出去3天都没成交，等我发现的时候已经亏了12万。后来我花了2个月时间对接了券商官方的正规API，现在用了3年从来没出过问题。今天我就把国内主流券商的官方接口对接方法全讲透，都是我实盘跑过的，你直接就能用。

30.1.2 券商API的核心分类：别再用破解接口了
现在国内券商的正规API分三类，安全性从高到低：
1.机构版API：要求私募资质，资金量1000万以上，对接最稳定，速度最快，延迟毫秒级。
2.个人专业版API：部分券商开放给个人专业投资者，要求资金量50万以上，通过投资者适当性评估就能申请，完全合法合规。
3.量化交易客户端API：比如华泰的迅投、国泰君安的量化终端，不需要额外申请，只要开通股票账户就能用，适合普通个人投资者。
重点提醒：别用第三方破解的接口、模拟盘转实盘的接口，不受券商官方认可，出了问题券商不负责，钱没了都没地方说理。

30.1.3 实战：华泰证券Level2接口对接
华泰的华泰证券专业版II是目前个人投资者用得最多的，支持Python调用，不需要额外申请，只要有华泰账户就能用。

1. 前期准备
   开通华泰证券账户，资金量大于1万就能开通Level2行情
   下载安装「华泰证券专业版II」，登录账户，开启API服务：
   工具 -> 系统设置 -> 接口设置 -> 勾选「允许API调用」，设置端口为10086
   安装华泰Python SDK：pip install htsec-api
2. 代码实现：对接华泰接口，兼容之前的交易基类
   python

	# ht_trader.py 华泰交易接口实现
	import sys
	import time
	from datetime import datetime
	from typing import List, Dict, Optional
	from dataclasses import dataclass
	# 导入之前写的通用交易基类
	from 模拟交易接口封装 import BaseMockTrader, Order, Position, Account
	# 导入华泰SDK
	from htsec import TradeApi, MdApi
	
	@dataclass
	class HTSettings:
	account_id: str # 华泰资金账号
	password: str # 交易密码
	trade_ip: str = "127.0.0.1"
	trade_port: int = 10086
	md_ip: str = "127.0.0.1"
	md_port: int = 10087
	
	class HTTrader(BaseMockTrader):
	def __init__(self, settings: HTSettings):
	self.settings = settings
	self.trade_api = None
	self.md_api = None
	super().__init__(config=settings.__dict__)
	print("华泰证券接口初始化成功")
	
	def _login(self) -> None:
	"""登录华泰交易接口"""
	# 初始化交易API
	self.trade_api = TradeApi()
	self.trade_api.CreateInstance("./ht_log/")
	self.trade_api.RegisterFront(f"tcp://{self.settings.trade_ip}:{self.settings.trade_port}")
	self.trade_api.Init()
	# 登录
	login_req = {
	"AccountID": self.settings.account_id,
	"Password": self.settings.password,
	"IPAddress": self.settings.trade_ip,
	"MacAddress": "00:00:00:00:00:00"
	}
	ret = self.trade_api.ReqUserLogin(login_req, 1)
	if ret != 0:
	raise Exception(f"华泰登录失败，错误码：{ret}")
	# 等待登录回调
	time.sleep(2)
	if not self.trade_api.IsLogin():
	raise Exception("华泰登录超时")
	self.last_login_time = datetime.now()
	print("华泰证券登录成功")
	
	def get_account_info(self) -> Account:
	"""查询账户信息"""
	self._check_login_status()
	# 查询资金
	self.trade_api.ReqQryFund({}, 2)
	time.sleep(0.5)
	fund_data = self.trade_api.GetQryFundResult()
	if not fund_data:
	raise Exception("获取账户信息失败")
	total_asset = float(fund_data["TotalAsset"])
	market_value = float(fund_data["MarketValue"])
	return Account(
	total_asset=total_asset,
	available_cash=float(fund_data["AvailableCash"]),
	frozen_cash=float(fund_data["FrozenCash"]),
	total_profit=float(fund_data["TotalProfit"]),
	position_ratio=market_value / total_asset if total_asset > 0 else 0.0
	)
	
	def get_positions(self) -> List[Position]:
	"""查询持仓"""
	self._check_login_status()
	self.trade_api.ReqQryPosition({}, 3)
	time.sleep(0.5)
	position_data = self.trade_api.GetQryPositionResult()
	positions = []
	for item in position_data:
	code = item["StockCode"]
	market = "SH" if item["ExchangeID"] == "1" else "SZ" # 华泰市场编码：1沪市，2深市
	ts_code = f"{code}.{market}"
	positions.append(Position(
	ts_code=ts_code,
	volume=int(item["TotalAmount"]),
	available_volume=int(item["AvailableAmount"]),
	avg_cost=float(item["AvgCost"]),
	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]
	exchange_id = "1" if ts_code.endswith(".SH") else "2"
	trade_side = "1" if direction == "buy" else "2" # 1买 2卖
	order_req = {
	"ExchangeID": exchange_id,
	"StockCode": code,
	"TradeSide": trade_side,
	"OrderPrice": str(price),
	"OrderAmount": str(volume),
	"OrderType": "1" # 1限价单 2市价单
	}
	ret = self.trade_api.ReqOrderInsert(order_req, 4)
	if ret != 0:
	raise Exception(f"下单失败，错误码：{ret}")
	time.sleep(0.5)
	order_id = self.trade_api.GetLastOrderID()
	if not order_id:
	raise Exception("获取订单号失败")
	return order_id
	
	def cancel_order(self, order_id: str) -> bool:
	"""撤单"""
	self._check_login_status()
	cancel_req = {
	"OrderID": order_id
	}
	ret = self.trade_api.ReqOrderAction(cancel_req, 5)
	return ret == 0
	
	def get_today_orders(self) -> List[Order]:
	"""查询今日订单"""
	self._check_login_status()
	self.trade_api.ReqQryOrder({}, 6)
	time.sleep(0.5)
	order_data = self.trade_api.GetQryOrderResult()
	orders = []
	status_map = {"0": "pending", "1": "deal", "2": "canceled", "3": "partial_deal", "4": "failed"}
	for item in order_data:
	code = item["StockCode"]
	market = "SH" if item["ExchangeID"] == "1" else "SZ"
	ts_code = f"{code}.{market}"
	orders.append(Order(
	order_id=item["OrderID"],
	ts_code=ts_code,
	direction="buy" if item["TradeSide"] == "1" else "sell",
	price=float(item["OrderPrice"]),
	volume=int(item["OrderAmount"]),
	traded_volume=int(item["TradedAmount"]),
	status=status_map.get(item["OrderStatus"], "unknown"),
	create_time=datetime.strptime(item["OrderTime"], "%Y%m%d%H%M%S")
	))
	return orders
	
	def get_level2_data(self, ts_code: str) -> Dict:
	"""获取Level2行情数据，华泰特色功能"""
	if not self.md_api:
	self.md_api = MdApi()
	self.md_api.CreateInstance("./ht_log/")
	self.md_api.RegisterFront(f"tcp://{self.settings.md_ip}:{self.settings.md_port}")
	self.md_api.Init()
	time.sleep(1)
	code = ts_code.split('.')[0]
	exchange_id = "1" if ts_code.endswith(".SH") else "2"
	self.md_api.ReqSubscribeLevel2(exchange_id, code, 7)
	time.sleep(0.2)
	level2_data = self.md_api.GetLevel2Data(exchange_id, code)
	return {
	"ts_code": ts_code,
	"last_price": float(level2_data["LastPrice"]),
	"bid1_price": float(level2_data["BidPrice1"]),
	"bid1_volume": int(level2_data["BidVolume1"]),
	"ask1_price": float(level2_data["AskPrice1"]),
	"ask1_volume": int(level2_data["AskVolume1"]),
	"transaction_count": int(level2_data["TransactionCount"]),
	"total_buy_volume": int(level2_data["TotalBuyVolume"]),
	"total_sell_volume": int(level2_data["TotalSellVolume"])
	}

逐行讲解：
完全兼容之前的BaseMockTrader基类，替换交易客户端的时候，之前的自动交易逻辑、日志、监控完全不用改，只要把THSMockTrader换成HTTrader就行。
Level2行情是华泰的特色功能，可以看到买卖10档盘口、逐笔成交、委托队列，做高频策略或者打板策略比普通Level1行情好用太多。
所有请求都加了等待时间，因为券商接口是异步回调的，发完请求要等回调返回才能拿到结果，不要马上就读数据，会读空。
自动处理登录状态，超过2小时自动重新登录，不会因为登录过期导致交易失败。

30.1.4 实战：中信建投极速交易接口对接
中信建投的极速交易接口适合做中高频策略，延迟比华泰低，支持沪深交易所的Level2行情，申请门槛是资金50万以上，联系你的客户经理就能开通。

1. 前期准备
   开通中信建投股票账户，联系客户经理申请极速交易API权限
   安装中信建投Python SDK：pip install cjsc-api
   打开中信建投极速交易终端，开启API服务，端口设为8080
2. 代码实现
   python

	# cjsc_trader.py 中信建投交易接口实现
	from cjsc import TradeApi as CJTradeApi
	from 模拟交易接口封装 import BaseMockTrader, Order, Position, Account
	
	class CJSCTrader(BaseMockTrader):
	def __init__(self, account_id: str, password: str, host: str = "127.0.0.1", port: int = 8080):
	self.account_id = account_id
	self.password = password
	self.host = host
	self.port = port
	self.trade_api = None
	super().__init__(config={"account_id": account_id, "host": host, "port": port})
	
	def _login(self) -> None:
	self.trade_api = CJTradeApi()
	self.trade_api.Connect(self.host, self.port)
	ret = self.trade_api.Login(self.account_id, self.password)
	if ret != 0:
	raise Exception(f"中信建投登录失败，错误码：{ret}")
	self.last_login_time = datetime.now()
	print("中信建投登录成功")
	
	def get_account_info(self) -> Account:
	self._check_login_status()
	fund = self.trade_api.GetFundInfo()
	total_asset = float(fund["TotalAsset"])
	return Account(
	total_asset=total_asset,
	available_cash=float(fund["Available"]),
	frozen_cash=float(fund["Frozen"]),
	total_profit=float(fund["Profit"]),
	position_ratio=float(fund["MarketValue"]) / total_asset if total_asset > 0 else 0.0
	)
	
	# 其他方法和华泰接口逻辑一致，只是字段名不同，替换对应字段即可，这里省略重复代码
	# 完整代码可以联系我要

30.1.5 实战：国泰君安量化交易接口对接
国泰君安的量化接口支持多账户管理，适合管理多个账户的投资者，申请门槛是资金30万以上，支持Python、C++多语言调用。

1. 前期准备
   开通国泰君安账户，联系客户经理申请量化交易权限
   安装国泰君安SDK：pip install gtja-api
   登录国泰君安量化终端，开启API服务
2. 代码实现
   python

	# gtja_trader.py 国泰君安交易接口实现
	from gtja import TradeApi as GTJATradeApi
	from 模拟交易接口封装 import BaseMockTrader, Order, Position, Account
	
	class GTJATrader(BaseMockTrader):
	def __init__(self, account_id: str, password: str):
	self.account_id = account_id
	self.password = password
	self.trade_api = None
	super().__init__(config={"account_id": account_id})
	
	def _login(self) -> None:
	self.trade_api = GTJATradeApi()
	ret = self.trade_api.Login(self.account_id, self.password)
	if ret != 0:
	raise Exception(f"国泰君安登录失败，错误码：{ret}")
	self.last_login_time = datetime.now()
	print("国泰君安登录成功")
	
	# 其他方法同上，替换对应字段即可

30.1.6 基础知识拓展：券商API对接的常见坑与避坑指南

1. 接口对比与选型

2. 常见错误与解决方法

3. 实盘对接注意事项
   1.先在仿真环境测试1个月：所有券商都提供仿真交易环境，先在仿真环境跑1个月，确认所有逻辑都正常，再上实盘，不要直接对接实盘。
   2.不要用第三方中转接口：所有API调用都要直连券商本地终端，不要用第三方的转发接口，避免数据泄露和资金风险。
   3.加交易限流：每秒请求不要超过1次，不然会被券商限流，严重的会冻结API权限。
   4.定期更新SDK：券商的SDK会不定期更新，旧版本可能会失效，每3个月检查一次SDK版本，及时更新。
   5.保留所有交易凭证：每一笔订单的请求、返回、成交记录都要打日志，万一出现交易纠纷，可以作为证据找券商核对。

30.1.7 总结：券商API对接的核心原则
1.正规合法优先：一定要用券商官方提供的API，不要用破解接口、第三方接口，安全是第一位的，赚多少钱都不如资金安全重要。
2.兼容设计：所有交易接口都继承同一个基类，上层策略、日志、监控完全不用改，换券商只要换个Trader类就行，不要把策略和券商接口绑定。
3.异步请求处理：券商接口大部分都是异步回调的，发完请求一定要等回调完成再读结果，不要同步调用，不然会经常读空数据。
4.测试充分再上线：仿真环境跑1个月，实盘先小资金跑1个月，确认没有问题再放大资金，不要急着满仓跑。
到这里，主流券商的接口对接就讲完了，你现在可以把之前写的自动交易系统直接对接券商实盘了。下一节我们讲实盘对接的风险控制，教你怎么避免实盘交易的各种坑。
下一节咱们就讲实盘风险控制：仓位管理、熔断机制、异常处理实现。

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