量化研究---bigquant策略交易api研究
api接口来平台的代码整理，原理是读取bigquant的模拟测试信号，下单，可以完美的对接qmt交易，我优化了交易api的部分内容
我开发对接qmt的交易系统
看api源代码
源代码
# 导入系统包
import os
import json
import requests
from typing import Dict, List, Optional, Tuple
from datetime import datetime
import pandas as pd
class xg_bigqunt_trader(object):"""小果宽邦bigquant交易信号api来自网页改写"""HOST = "https://bigquant.com"ACCESS_KEY_HEADER_NAME: str = "X-BigQuant-Access-Key"ACCESS_SIGNATURE_HEADER_NAME: str = "X-BigQuant-Signature"ACCESS_TIMESTAMP_HEADER_NAME: str = "X-BigQuant-Timestamp"def __init__(self, access_key: str, secret_key: str):self._access_key = access_keyself._secret_key = secret_keyself.account_type_dict ={}self.account_id_dict = {}self.tradingapiserver_host = os.getenv("TRADINGAPISERVER_HOST")self.debug = 1def query_papertrading_info(self,strategy_id='ba730553-e42a-4105-84ec-a3c47c7ac7f6') -> Optional[Dict]:"""获取模拟交易基本信息example data:{'id': 'aec3e784-0724-4d7d-8917-16461b8c5553', 'created_at': '2023-12-07T17:52:43.232000+08:00', 'updated_at': '2023-12-07T17:53:08.416973+08:00','space_id': '00000000-0000-0000-0000-000000000000', 'creator': '6d0130a2-3d94-4f38-97a2-14e1dc9391a4','strategy_name': 'StockRanker-DAI策略-20231206112622', 'strategy_type': '0', 'status': 1, 'account_id': {'10020': '0'},'benchmark_instrument': '000300.SH', 'first_benchmark_index': 3572.36, 'first_trading_day': '2023-12-07', 'frequency': '1d','extension': {}, 'source': 'saas', 'strategy_params': {'volume_limit': 1, 'order_price_field_buy': 'open', 'order_price_field_sell': 'close'}"""url = f"{self.tradingapiserver_host}/strategies/{strategy_id}"params = {'page': 1, 'size': 10}if self.tradingapiserver_host:url = f"{self.tradingapiserver_host}/strategies/{strategy_id}"headers = Noneelse:url, headers = self.signature_headers(self.HOST, f"/bigapis/trading/v1/strategies/{strategy_id}", self._access_key, self._secret_key, params=params)resp = requests.get(url=url, headers=headers, params=params)resp_data = resp.json()if self.debug:#print(f"query_papertrading_info: resp_data={resp_data}")passif resp_data["code"] != 0:raise Exception(resp_data["message"])ret = resp_data.get("data")account_id_dict = ret["account_id"]account_type = list(account_id_dict.values())[0]account_id = list(account_id_dict.keys())[0]self.account_type_dict[strategy_id]=account_typeself.account_id_dict[strategy_id]=account_id#print(f"query_papertrading_info: got account_id='{account_type}',\"{account_id}\" by strategy_id={strategy_id}")return retdef query_papertrading_cash(self,strategy_id='ba730553-e42a-4105-84ec-a3c47c7ac7f6',trading_day='2025-04-17') -> Optional[Dict]:"""获取模拟交易资金信息example data:{'account_type': '0', 'account_id': '10020', 'trading_day': '2023-06-19', 'currency': 'CNY', 'balance': 109016.71000000006,'available': 109016.71000000006, 'frozen_cash': 0.0, 'total_margin': 0.0, 'total_market_value': 407791.0, 'portfolio_value': 516807.71,'pre_balance': 101875.81000000006, 'positions_pnl': 12350.64, 'capital_changed': 0.0, 'total_capital_changed': 0.0}"""if strategy_id not in list(set(self.account_type_dict.keys())):self.query_papertrading_info(strategy_id=strategy_id)else:passaccount_type=self.account_type_dict[strategy_id]account_id=self.account_id_dict[strategy_id]constraints = {"account_type": account_type, "trading_day__gte": trading_day, "trading_day__lte": trading_day}params = {'constraints': json.dumps(constraints), 'page': 1, 'size': 10}if self.tradingapiserver_host:url, headers = f"{self.tradingapiserver_host}/accounts/{account_id}/cash", Noneelse:url, headers = self.signature_headers(self.HOST, f"/bigapis/trading/v1/accounts/{account_id}/cash", self._access_key, self._secret_key, params=params)resp = requests.get(url=url, headers=headers, params=params)resp_data = resp.json()if self.debug:#print(f"query_papertrading_cash: resp_data={resp_data}")passif resp_data["code"] != 0:raise Exception(resp_data["message"])datas = resp_data.get("data").get("items")if datas:ret = datas[0]if "created_at" in ret:del ret["created_at"]if "id" in ret:del ret["id"]if "updated_at" in ret:del ret["updated_at"]return retelse:return {}def query_papertrading_positions(self,strategy_id='ba730553-e42a-4105-84ec-a3c47c7ac7f6', trading_day='2025-04-17') -> List[Dict]:"""获取模拟交易某日的持仓列表example data:[{'account_type': '0', 'account_id': '10020', 'trading_day': '2023-06-19', 'exchange': 'SZ', 'instrument': '000005.SZ','name': 'ST星源', 'posi_direction': '1', 'current_qty': 17800, 'available_qty': 17800, 'today_qty': 0, 'today_available_qty': 0,'cost_price': 1.17, 'last_price': 1.22, 'market_value': 21716.0, 'margin': 0.0, 'position_pnl': 887.92, 'hedge_flag': '1','sum_buy_value': 20826.0, 'sum_sell_value': 0.0, 'commission': 2.08, 'dividend_qty': 0, 'dividend_cash': 0.0,'open_date': '2023-06-16', 'open_price': 1.17, 'settlement_price': 0.0, 'hold_days': 2}]"""if strategy_id not in list(set(self.account_type_dict.keys())):self.query_papertrading_info(strategy_id=strategy_id)else:passaccount_type=self.account_type_dict[strategy_id]account_id=self.account_id_dict[strategy_id]constraints = {"account_type": account_type, "trading_day__gte": trading_day, "trading_day__lte": trading_day}params = {'constraints': json.dumps(constraints), 'page': 1, 'size': 3000}if self.tradingapiserver_host:url, headers = f"{self.tradingapiserver_host}/account_positions/{account_id}/positions", Noneelse:url, headers = self.signature_headers(self.HOST, f"/bigapis/trading/v1/account_positions/{account_id}/positions", self._access_key, self._secret_key, params=params)resp = requests.get(url=url, headers=headers, params=params)resp_data = resp.json()if resp_data["code"] != 0:raise Exception(resp_data["message"])list_positions = []datas = resp_data.get("data").get("items")if datas:for pos_data in datas:if "created_at" in pos_data:del pos_data["created_at"]if "id" in pos_data:del pos_data["id"]if "updated_at" in pos_data:del pos_data["updated_at"]list_positions.append(pos_data)else:list_positions=list_positionsif len(list_positions):df=pd.DataFrame(list_positions)df=df[['trading_day','instrument','name',"cost_price","today_qty",'hold_days',"open_price","market_value","position_pnl","profit_ratio"]]df.columns=['交易日','证券代码','名称',"成本",'数量','持有天数',"收盘价",'市值','浮动盈亏','收益率%']df['收益率%']=df['收益率%']*100else:df=pd.DataFrame()return dfdef query_papertrading_positions_all(self,strategy_id='ba730553-e42a-4105-84ec-a3c47c7ac7f6') -> List[Dict]:"""获取全部历史的持股example data:[{'account_type': '0', 'account_id': '10020', 'trading_day': '2023-06-19', 'exchange': 'SZ', 'instrument': '000005.SZ','name': 'ST星源', 'posi_direction': '1', 'current_qty': 17800, 'available_qty': 17800, 'today_qty': 0, 'today_available_qty': 0,'cost_price': 1.17, 'last_price': 1.22, 'market_value': 21716.0, 'margin': 0.0, 'position_pnl': 887.92, 'hedge_flag': '1','sum_buy_value': 20826.0, 'sum_sell_value': 0.0, 'commission': 2.08, 'dividend_qty': 0, 'dividend_cash': 0.0,'open_date': '2023-06-16', 'open_price': 1.17, 'settlement_price': 0.0, 'hold_days': 2}]"""if strategy_id not in list(set(self.account_type_dict.keys())):self.query_papertrading_info(strategy_id=strategy_id)else:passaccount_type=self.account_type_dict[strategy_id]account_id=self.account_id_dict[strategy_id]constraints = {"account_type": account_type}params = {'constraints': json.dumps(constraints), 'page': 1, 'size': 3000}if self.tradingapiserver_host:url, headers = f"{self.tradingapiserver_host}/account_positions/{account_id}/positions", Noneelse:url, headers = self.signature_headers(self.HOST, f"/bigapis/trading/v1/account_positions/{account_id}/positions", self._access_key, self._secret_key, params=params)resp = requests.get(url=url, headers=headers, params=params)resp_data = resp.json()if resp_data["code"] != 0:raise Exception(resp_data["message"])list_positions = []datas = resp_data.get("data").get("items")if datas:for pos_data in datas:if "created_at" in pos_data:del pos_data["created_at"]if "id" in pos_data:del pos_data["id"]if "updated_at" in pos_data:del pos_data["updated_at"]list_positions.append(pos_data)else:list_positions=list_positionsif len(list_positions):df=pd.DataFrame(list_positions)df=df[['trading_day','instrument','name',"cost_price","today_qty",'hold_days',"open_price","market_value","position_pnl","profit_ratio"]]df.columns=['交易日','证券代码','名称',"成本",'数量','持有天数',"收盘价",'市值','浮动盈亏','收益率%']df['收益率%']=df['收益率%']*100else:df=pd.DataFrame()return dfdef query_papertrading_planned_orders(self, strategy_id='ba730553-e42a-4105-84ec-a3c47c7ac7f6', trading_day='2025-04-17') -> List[Dict]:"""获取模拟交易某日的信号列表example data:[{'creator': '6d0130a2-3d94-4f38-97a2-14e1dc9391a4', 'planned_order_id': '145909290', 'strategy_id': 'aec3e784-0724-4d7d-8917-16461b8c5553','account_type': '0', 'account_id': '10020', 'trading_day': '2023-06-19', 'order_dt': '2023-06-19T15:00:00+08:00', 'exchange': 'SH','instrument': '600767.SH', 'name': '*ST运盛', 'direction': '2', 'offset_flag': '1', 'original_order_qty': 251600, 'order_qty': 251600,'order_price': 0.42, 'order_type': 'U', 'order_status': 10, 'status_msg': 'Generated', 'order_params': None, 'order_placed_dt': None,'order_key': '', 'entrust_no': '', 'algo_order_id': 0, 'stop_loss_price': 0.0, 'stop_profit_price': 0.0}]"""if strategy_id not in list(set(self.account_type_dict.keys())):self.query_papertrading_info(strategy_id=strategy_id)else:passaccount_type=self.account_type_dict[strategy_id]account_id=self.account_id_dict[strategy_id]constraints = {"account_type": account_type,"account_id": account_id,"trading_day__gte": trading_day,"trading_day__lte": trading_day}params = {'constraints': json.dumps(constraints), 'page': 1, 'size': 3000}if self.tradingapiserver_host:url, headers = f"{self.tradingapiserver_host}/planned_order", Noneelse:url, headers = self.signature_headers(self.HOST, f"/bigapis/trading/v1/planned_order", self._access_key, self._secret_key, params=params)resp = requests.get(url=url, headers=headers, params=params)resp_data = resp.json()if resp_data["code"] != 0:raise Exception(resp_data["message"])list_planned_orders: List[Dict] = []######### For test mock planned orders"""_planned_order = {'planned_order_id': '1','strategy_id': self._strategy_id,'account_type': '0', 'account_id': '10000','trading_day': trading_day,'order_dt': ' '.join([trading_day, "09:31:00"]),'exchange': 'SH','instrument': '600900.SH','name': '长江电力','direction': '1','offset_flag': '0','original_order_qty': 600,'order_qty': 600,'order_price': 23.42,'order_type': 'U','order_status': 10,'status_msg': 'Generated'}list_planned_orders.append(_planned_order)return list_planned_orders"""########datas = resp_data.get("data").get("items")if datas:for planned_order in datas:if "created_at" in planned_order:del planned_order["created_at"]if "id" in planned_order:del planned_order["id"]if "updated_at" in planned_order:del planned_order["updated_at"]list_planned_orders.append(planned_order)else:list_planned_orders=list_planned_ordersif len(list_planned_orders):df=pd.DataFrame(list_planned_orders)df=df[['trading_day','instrument','name','original_order_qty','order_price','direction']]df.columns=['交易日','证券代码','名称','数量','价格','交易类型']df['交易类型']=df['交易类型'].apply(lambda x:'买' if x=='1' else '卖')else:df=pd.DataFrame()return dfdef query_papertrading_planned_orders_all(self, strategy_id='ba730553-e42a-4105-84ec-a3c47c7ac7f6') -> List[Dict]:"""获取模拟交易全部信号策略example data:[{'creator': '6d0130a2-3d94-4f38-97a2-14e1dc9391a4', 'planned_order_id': '145909290', 'strategy_id': 'aec3e784-0724-4d7d-8917-16461b8c5553','account_type': '0', 'account_id': '10020', 'trading_day': '2023-06-19', 'order_dt': '2023-06-19T15:00:00+08:00', 'exchange': 'SH','instrument': '600767.SH', 'name': '*ST运盛', 'direction': '2', 'offset_flag': '1', 'original_order_qty': 251600, 'order_qty': 251600,'order_price': 0.42, 'order_type': 'U', 'order_status': 10, 'status_msg': 'Generated', 'order_params': None, 'order_placed_dt': None,'order_key': '', 'entrust_no': '', 'algo_order_id': 0, 'stop_loss_price': 0.0, 'stop_profit_price': 0.0}]"""if strategy_id not in list(set(self.account_type_dict.keys())):self.query_papertrading_info(strategy_id=strategy_id)else:passaccount_type=self.account_type_dict[strategy_id]account_id=self.account_id_dict[strategy_id]constraints = {"account_type": account_type,"account_id": account_id,}params = {'constraints': json.dumps(constraints), 'page': 1, 'size': 3000}if self.tradingapiserver_host:url, headers = f"{self.tradingapiserver_host}/planned_order", Noneelse:url, headers = self.signature_headers(self.HOST, f"/bigapis/trading/v1/planned_order", self._access_key, self._secret_key, params=params)resp = requests.get(url=url, headers=headers, params=params)resp_data = resp.json()if resp_data["code"] != 0:raise Exception(resp_data["message"])list_planned_orders: List[Dict] = []######### For test mock planned orders"""_planned_order = {'planned_order_id': '1','strategy_id': self._strategy_id,'account_type': '0', 'account_id': '10000','trading_day': trading_day,'order_dt': ' '.join([trading_day, "09:31:00"]),'exchange': 'SH','instrument': '600900.SH','name': '长江电力','direction': '1','offset_flag': '0','original_order_qty': 600,'order_qty': 600,'order_price': 23.42,'order_type': 'U','order_status': 10,'status_msg': 'Generated'}list_planned_orders.append(_planned_order)return list_planned_orders"""########datas = resp_data.get("data").get("items")if datas:for planned_order in datas:if "created_at" in planned_order:del planned_order["created_at"]if "id" in planned_order:del planned_order["id"]if "updated_at" in planned_order:del planned_order["updated_at"]list_planned_orders.append(planned_order)else:list_planned_orders=list_planned_ordersif len(list_planned_orders):df=pd.DataFrame(list_planned_orders)df=df[['trading_day','instrument','name','original_order_qty','order_price','direction']]df.columns=['交易日','证券代码','名称','数量','价格','交易类型']df['交易类型']=df['交易类型'].apply(lambda x:'买' if x=='1' else '卖')else:df=pd.DataFrame()return dfdef signature_headers(self,host: str,path: str,access_key: str,secret_key: str,body: bytes = b"",headers: dict = {},params: dict = None,) -> Tuple[str, dict]:"""采用aksk给headers添加签名.Args:host (str): host.path (str): 路径.access_key (str): ak.secret_key (str): sk.body (bytes): 请求负载.headers (dict, optional): 请求头.Returns:Tuple[str, dict]: url, 请求头."""import hmacimport timefrom urllib.parse import quoteurl = f"{host}{path}"if params:url = url + "?" + quote("&".join([f"{k}={v}" for k, v in params.items()]))path_encode = path.encode()timestamp = int(time.time() * 1000)timestamp = str(timestamp)# 参与签名的消息msg = path_encode + body + timestamp.encode()signature = hmac.new(secret_key.encode(), msg=msg, digestmod="SHA256").hexdigest()# 给headers这是签名和时间戳headers[self.ACCESS_KEY_HEADER_NAME] = access_keyheaders[self.ACCESS_SIGNATURE_HEADER_NAME] = signatureheaders[self.ACCESS_TIMESTAMP_HEADER_NAME] = timestampreturn url, headers
if __name__ == "__main__":access_key = "J0kem11dvUy1"secret_key = "wnsvthHmHVuxHMStmNvGJyNxqR7lB0Eog7SfWDCkiIPiHfA3KwtCwoOebhqAUTiq"strategy_id = "4eeee2eb-fc9b-4b25-8442-d99c70d9ce5f"trading_day = '2025-05-23' # 假设交易日为当天# 初始化 API 对象api = xg_bigqunt_trader(access_key, secret_key)#先调这个函数获取参数账户类型#账户信息df=api.query_papertrading_info()print(df)#账户现金df=api.query_papertrading_cash()print(df)#持股df=api.query_papertrading_positions(strategy_id,trading_day)print(df)#持股全部df=api.query_papertrading_positions(strategy_id)print(df)#交易信号df=api.query_papertrading_planned_orders(strategy_id,trading_day)print(df)#全部交易信号df=api.query_papertrading_planned_orders_all(strategy_id)print(df)里面有3个参数，公钥，私钥，策略id
1公钥，私钥获取
点击登录的主页
2点击点击访问凭证
3点击新建凭证
点击建立
点击复制私钥 比如RnnA00dqd13r6Nl0e696LWOFG3DYLxwSql6RKpJDecosstyuW87EoWr26K7ma5tx
我们就完成公钥
私钥RnnA00dqd13r6Nl0e696LWOFG3DYLxwSql6RKpJDecosstyuW87EoWr26K7ma5tx设置hhuo
获取策略id 点击策略模拟，我的策略
随便点击一个策略
上面部分就是策略id 4eeee2eb-fc9b-4b25-8442-d99c70d9ce5f
读取策略数据
和网页一模一样
获取的内容
{'id': 'ba730553-e42a-4105-84ec-a3c47c7ac7f6', 'created_at': '2025-04-17T11:51:25.685860+08:00', 'updated_at': '2025-04-22T15:00:55.391398+08:00', 'space_id': '00000000-0000-0000-0000-000000000000', 'creator': 'e8dfa494-7faa-470c-8b07-c9ba86d68be5', 'strategy_name': '量化选股与动态调整', 'strategy_type': '0', 'status': 2, 'account_id': {'20ba0f0a-18ec-4a9f-b210-ce9062d01bc8': '0'}, 'benchmark_instrument': '000300.SH', 'first_benchmark_index': 3772.8204, 'first_trading_day': '2025-04-16', 'frequency': '1d', 'extension': {}, 'source': 'saas', 'strategy_params': {'volume_limit': 1, 'order_price_field_buy': 'open', 'order_price_field_sell': 'close'}, 'push_conf': {}, 'highlights': {}, 'data': {}}
{'account_type': '0', 'account_id': '20ba0f0a-18ec-4a9f-b210-ce9062d01bc8', 'trading_day': '2025-04-17', 'currency': 'CNY', 'balance': 18846.75999999995, 'available': 18846.75999999995, 'frozen_cash': 0.0, 'total_margin': 0.0, 'total_market_value': 983172.0, 'portfolio_value': 1002018.76, 'pre_balance': 1000000.0, 'positions_pnl': 2018.76, 'capital_changed': 0.0, 'total_capital_changed': 0.0, 'premium': 0.0, 'commission': 294.24, 'realized_pnl': 0.0}交易日 证券代码 名称 成本 数量 持有天数 收盘价 市值 浮动盈亏 收益率%
0 2025-05-23 300204.SZ 舒泰神 14.094227 0 2 14.09 0.0 228408.21 17.064
1 2025-05-23 301571.SZ 国科天成 50.885259 1700 1 50.87 85051.0 -1453.94 -1.681
Empty DataFrame
Columns: []
Index: []交易日 证券代码 名称 数量 价格 交易类型
0 2025-05-23 300204.SZ 舒泰神 95000 14.78 卖
1 2025-05-23 301571.SZ 国科天成 1700 53.69 买交易日 证券代码 名称 数量 价格 交易类型
0 2025-05-06 300005.SZ 探路者 98000 10.15 买
1 2025-05-07 300005.SZ 探路者 90900 10.70 卖
2 2025-05-08 300281.SZ 金明精机 110300 8.70 买
3 2025-05-09 300281.SZ 金明精机 107100 8.60 卖
4 2025-05-09 300032.SZ 金龙机电 100 5.92 买
5 2025-05-12 300032.SZ 金龙机电 100 5.17 卖
6 2025-05-12 301317.SZ 鑫磊股份 25800 32.70 买
7 2025-05-13 301317.SZ 鑫磊股份 25800 31.80 卖
8 2025-05-13 300581.SZ 晨曦航空 6000 18.06 买
9 2025-05-14 300581.SZ 晨曦航空 6000 16.47 卖
10 2025-05-14 300946.SZ 恒而达 16600 58.80 买
11 2025-05-15 300946.SZ 恒而达 15100 70.56 卖
12 2025-05-15 300399.SZ 天利科技 3700 25.44 买
13 2025-05-16 300399.SZ 天利科技 3700 25.55 卖
14 2025-05-16 301089.SZ 拓新药业 34300 31.48 买
15 2025-05-19 301089.SZ 拓新药业 33700 37.78 卖
16 2025-05-19 301488.SZ 豪恩汽电 1300 73.80 买
17 2025-05-20 301488.SZ 豪恩汽电 1100 80.25 卖
18 2025-05-20 301459.SZ 丰茂股份 19600 65.75 买
19 2025-05-21 301459.SZ 丰茂股份 19400 70.30 卖
20 2025-05-21 300723.SZ 一品红 1700 47.53 买
21 2025-05-22 300723.SZ 一品红 1700 50.60 卖
22 2025-05-22 300204.SZ 舒泰神 95000 14.15 买
23 2025-05-23 300204.SZ 舒泰神 95000 14.78 卖
24 2025-05-23 301571.SZ 国科天成 1700 53.69 买
25 2025-05-26 301571.SZ 国科天成 1700 50.03 卖
26 2025-05-26 301209.SZ 联合化学 12600 124.15 买
PS C:\Users\lxg123456\Desktop\小果宽邦成交交易系统> 交易api对接的源代码，我改进了不少内容
# 导入系统包
import os
import json
import requests
from typing import Dict, List, Optional, Tuple
from datetime import datetime
import pandas as pd
class xg_bigqunt_trader(object):"""小果宽邦bigquant交易信号api来自网页改写"""HOST = "https://bigquant.com"ACCESS_KEY_HEADER_NAME: str = "X-BigQuant-Access-Key"ACCESS_SIGNATURE_HEADER_NAME: str = "X-BigQuant-Signature"ACCESS_TIMESTAMP_HEADER_NAME: str = "X-BigQuant-Timestamp"def __init__(self, access_key: str, secret_key: str):self._access_key = access_keyself._secret_key = secret_keyself.account_type_dict ={}self.account_id_dict = {}self.tradingapiserver_host = os.getenv("TRADINGAPISERVER_HOST")self.debug = 1def query_papertrading_info(self,strategy_id='ba730553-e42a-4105-84ec-a3c47c7ac7f6') -> Optional[Dict]:"""获取模拟交易基本信息example data:{'id': 'aec3e784-0724-4d7d-8917-16461b8c5553', 'created_at': '2023-12-07T17:52:43.232000+08:00', 'updated_at': '2023-12-07T17:53:08.416973+08:00','space_id': '00000000-0000-0000-0000-000000000000', 'creator': '6d0130a2-3d94-4f38-97a2-14e1dc9391a4','strategy_name': 'StockRanker-DAI策略-20231206112622', 'strategy_type': '0', 'status': 1, 'account_id': {'10020': '0'},'benchmark_instrument': '000300.SH', 'first_benchmark_index': 3572.36, 'first_trading_day': '2023-12-07', 'frequency': '1d','extension': {}, 'source': 'saas', 'strategy_params': {'volume_limit': 1, 'order_price_field_buy': 'open', 'order_price_field_sell': 'close'}"""url = f"{self.tradingapiserver_host}/strategies/{strategy_id}"params = {'page': 1, 'size': 10}if self.tradingapiserver_host:url = f"{self.tradingapiserver_host}/strategies/{strategy_id}"headers = Noneelse:url, headers = self.signature_headers(self.HOST, f"/bigapis/trading/v1/strategies/{strategy_id}", self._access_key, self._secret_key, params=params)resp = requests.get(url=url, headers=headers, params=params)resp_data = resp.json()if self.debug:#print(f"query_papertrading_info: resp_data={resp_data}")passif resp_data["code"] != 0:raise Exception(resp_data["message"])ret = resp_data.get("data")account_id_dict = ret["account_id"]account_type = list(account_id_dict.values())[0]account_id = list(account_id_dict.keys())[0]self.account_type_dict[strategy_id]=account_typeself.account_id_dict[strategy_id]=account_id#print(f"query_papertrading_info: got account_id='{account_type}',\"{account_id}\" by strategy_id={strategy_id}")return retdef query_papertrading_cash(self,strategy_id='ba730553-e42a-4105-84ec-a3c47c7ac7f6',trading_day='2025-04-17') -> Optional[Dict]:"""获取模拟交易资金信息example data:{'account_type': '0', 'account_id': '10020', 'trading_day': '2023-06-19', 'currency': 'CNY', 'balance': 109016.71000000006,'available': 109016.71000000006, 'frozen_cash': 0.0, 'total_margin': 0.0, 'total_market_value': 407791.0, 'portfolio_value': 516807.71,'pre_balance': 101875.81000000006, 'positions_pnl': 12350.64, 'capital_changed': 0.0, 'total_capital_changed': 0.0}"""if strategy_id not in list(set(self.account_type_dict.keys())):self.query_papertrading_info(strategy_id=strategy_id)else:passaccount_type=self.account_type_dict[strategy_id]account_id=self.account_id_dict[strategy_id]constraints = {"account_type": account_type, "trading_day__gte": trading_day, "trading_day__lte": trading_day}params = {'constraints': json.dumps(constraints), 'page': 1, 'size': 10}if self.tradingapiserver_host:url, headers = f"{self.tradingapiserver_host}/accounts/{account_id}/cash", Noneelse:url, headers = self.signature_headers(self.HOST, f"/bigapis/trading/v1/accounts/{account_id}/cash", self._access_key, self._secret_key, params=params)resp = requests.get(url=url, headers=headers, params=params)resp_data = resp.json()if self.debug:#print(f"query_papertrading_cash: resp_data={resp_data}")passif resp_data["code"] != 0:raise Exception(resp_data["message"])datas = resp_data.get("data").get("items")if datas:ret = datas[0]if "created_at" in ret:del ret["created_at"]if "id" in ret:del ret["id"]if "updated_at" in ret:del ret["updated_at"]return retelse:return {}def query_papertrading_positions(self,strategy_id='ba730553-e42a-4105-84ec-a3c47c7ac7f6', trading_day='2025-04-17') -> List[Dict]:"""获取模拟交易某日的持仓列表example data:[{'account_type': '0', 'account_id': '10020', 'trading_day': '2023-06-19', 'exchange': 'SZ', 'instrument': '000005.SZ','name': 'ST星源', 'posi_direction': '1', 'current_qty': 17800, 'available_qty': 17800, 'today_qty': 0, 'today_available_qty': 0,'cost_price': 1.17, 'last_price': 1.22, 'market_value': 21716.0, 'margin': 0.0, 'position_pnl': 887.92, 'hedge_flag': '1','sum_buy_value': 20826.0, 'sum_sell_value': 0.0, 'commission': 2.08, 'dividend_qty': 0, 'dividend_cash': 0.0,'open_date': '2023-06-16', 'open_price': 1.17, 'settlement_price': 0.0, 'hold_days': 2}]"""if strategy_id not in list(set(self.account_type_dict.keys())):self.query_papertrading_info(strategy_id=strategy_id)else:passaccount_type=self.account_type_dict[strategy_id]account_id=self.account_id_dict[strategy_id]constraints = {"account_type": account_type, "trading_day__gte": trading_day, "trading_day__lte": trading_day}params = {'constraints': json.dumps(constraints), 'page': 1, 'size': 3000}if self.tradingapiserver_host:url, headers = f"{self.tradingapiserver_host}/account_positions/{account_id}/positions", Noneelse:url, headers = self.signature_headers(self.HOST, f"/bigapis/trading/v1/account_positions/{account_id}/positions", self._access_key, self._secret_key, params=params)resp = requests.get(url=url, headers=headers, params=params)resp_data = resp.json()if resp_data["code"] != 0:raise Exception(resp_data["message"])list_positions = []datas = resp_data.get("data").get("items")if datas:for pos_data in datas:if "created_at" in pos_data:del pos_data["created_at"]if "id" in pos_data:del pos_data["id"]if "updated_at" in pos_data:del pos_data["updated_at"]list_positions.append(pos_data)else:list_positions=list_positionsif len(list_positions):df=pd.DataFrame(list_positions)df=df[['trading_day','instrument','name',"cost_price","today_qty",'hold_days',"open_price","market_value","position_pnl","profit_ratio"]]df.columns=['交易日','证券代码','名称',"成本",'数量','持有天数',"收盘价",'市值','浮动盈亏','收益率%']df['收益率%']=df['收益率%']*100else:df=pd.DataFrame()return dfdef query_papertrading_positions_all(self,strategy_id='ba730553-e42a-4105-84ec-a3c47c7ac7f6') -> List[Dict]:"""获取全部历史的持股example data:[{'account_type': '0', 'account_id': '10020', 'trading_day': '2023-06-19', 'exchange': 'SZ', 'instrument': '000005.SZ','name': 'ST星源', 'posi_direction': '1', 'current_qty': 17800, 'available_qty': 17800, 'today_qty': 0, 'today_available_qty': 0,'cost_price': 1.17, 'last_price': 1.22, 'market_value': 21716.0, 'margin': 0.0, 'position_pnl': 887.92, 'hedge_flag': '1','sum_buy_value': 20826.0, 'sum_sell_value': 0.0, 'commission': 2.08, 'dividend_qty': 0, 'dividend_cash': 0.0,'open_date': '2023-06-16', 'open_price': 1.17, 'settlement_price': 0.0, 'hold_days': 2}]"""if strategy_id not in list(set(self.account_type_dict.keys())):self.query_papertrading_info(strategy_id=strategy_id)else:passaccount_type=self.account_type_dict[strategy_id]account_id=self.account_id_dict[strategy_id]constraints = {"account_type": account_type}params = {'constraints': json.dumps(constraints), 'page': 1, 'size': 3000}if self.tradingapiserver_host:url, headers = f"{self.tradingapiserver_host}/account_positions/{account_id}/positions", Noneelse:url, headers = self.signature_headers(self.HOST, f"/bigapis/trading/v1/account_positions/{account_id}/positions", self._access_key, self._secret_key, params=params)resp = requests.get(url=url, headers=headers, params=params)resp_data = resp.json()if resp_data["code"] != 0:raise Exception(resp_data["message"])list_positions = []datas = resp_data.get("data").get("items")if datas:for pos_data in datas:if "created_at" in pos_data:del pos_data["created_at"]if "id" in pos_data:del pos_data["id"]if "updated_at" in pos_data:del pos_data["updated_at"]list_positions.append(pos_data)else:list_positions=list_positionsif len(list_positions):df=pd.DataFrame(list_positions)df=df[['trading_day','instrument','name',"cost_price","today_qty",'hold_days',"open_price","market_value","position_pnl","profit_ratio"]]df.columns=['交易日','证券代码','名称',"成本",'数量','持有天数',"收盘价",'市值','浮动盈亏','收益率%']df['收益率%']=df['收益率%']*100else:df=pd.DataFrame()return dfdef query_papertrading_planned_orders(self, strategy_id='ba730553-e42a-4105-84ec-a3c47c7ac7f6', trading_day='2025-04-17') -> List[Dict]:"""获取模拟交易某日的信号列表example data:[{'creator': '6d0130a2-3d94-4f38-97a2-14e1dc9391a4', 'planned_order_id': '145909290', 'strategy_id': 'aec3e784-0724-4d7d-8917-16461b8c5553','account_type': '0', 'account_id': '10020', 'trading_day': '2023-06-19', 'order_dt': '2023-06-19T15:00:00+08:00', 'exchange': 'SH','instrument': '600767.SH', 'name': '*ST运盛', 'direction': '2', 'offset_flag': '1', 'original_order_qty': 251600, 'order_qty': 251600,'order_price': 0.42, 'order_type': 'U', 'order_status': 10, 'status_msg': 'Generated', 'order_params': None, 'order_placed_dt': None,'order_key': '', 'entrust_no': '', 'algo_order_id': 0, 'stop_loss_price': 0.0, 'stop_profit_price': 0.0}]"""if strategy_id not in list(set(self.account_type_dict.keys())):self.query_papertrading_info(strategy_id=strategy_id)else:passaccount_type=self.account_type_dict[strategy_id]account_id=self.account_id_dict[strategy_id]constraints = {"account_type": account_type,"account_id": account_id,"trading_day__gte": trading_day,"trading_day__lte": trading_day}params = {'constraints': json.dumps(constraints), 'page': 1, 'size': 3000}if self.tradingapiserver_host:url, headers = f"{self.tradingapiserver_host}/planned_order", Noneelse:url, headers = self.signature_headers(self.HOST, f"/bigapis/trading/v1/planned_order", self._access_key, self._secret_key, params=params)resp = requests.get(url=url, headers=headers, params=params)resp_data = resp.json()if resp_data["code"] != 0:raise Exception(resp_data["message"])list_planned_orders: List[Dict] = []######### For test mock planned orders"""_planned_order = {'planned_order_id': '1','strategy_id': self._strategy_id,'account_type': '0', 'account_id': '10000','trading_day': trading_day,'order_dt': ' '.join([trading_day, "09:31:00"]),'exchange': 'SH','instrument': '600900.SH','name': '长江电力','direction': '1','offset_flag': '0','original_order_qty': 600,'order_qty': 600,'order_price': 23.42,'order_type': 'U','order_status': 10,'status_msg': 'Generated'}list_planned_orders.append(_planned_order)return list_planned_orders"""########datas = resp_data.get("data").get("items")if datas:for planned_order in datas:if "created_at" in planned_order:del planned_order["created_at"]if "id" in planned_order:del planned_order["id"]if "updated_at" in planned_order:del planned_order["updated_at"]list_planned_orders.append(planned_order)else:list_planned_orders=list_planned_ordersif len(list_planned_orders):df=pd.DataFrame(list_planned_orders)df=df[['trading_day','instrument','name','original_order_qty','order_price','direction']]df.columns=['交易日','证券代码','名称','数量','价格','交易类型']df['交易类型']=df['交易类型'].apply(lambda x:'买' if x=='1' else '卖')else:df=pd.DataFrame()return dfdef query_papertrading_planned_orders_all(self, strategy_id='ba730553-e42a-4105-84ec-a3c47c7ac7f6') -> List[Dict]:"""获取模拟交易全部信号策略example data:[{'creator': '6d0130a2-3d94-4f38-97a2-14e1dc9391a4', 'planned_order_id': '145909290', 'strategy_id': 'aec3e784-0724-4d7d-8917-16461b8c5553','account_type': '0', 'account_id': '10020', 'trading_day': '2023-06-19', 'order_dt': '2023-06-19T15:00:00+08:00', 'exchange': 'SH','instrument': '600767.SH', 'name': '*ST运盛', 'direction': '2', 'offset_flag': '1', 'original_order_qty': 251600, 'order_qty': 251600,'order_price': 0.42, 'order_type': 'U', 'order_status': 10, 'status_msg': 'Generated', 'order_params': None, 'order_placed_dt': None,'order_key': '', 'entrust_no': '', 'algo_order_id': 0, 'stop_loss_price': 0.0, 'stop_profit_price': 0.0}]"""if strategy_id not in list(set(self.account_type_dict.keys())):self.query_papertrading_info(strategy_id=strategy_id)else:passaccount_type=self.account_type_dict[strategy_id]account_id=self.account_id_dict[strategy_id]constraints = {"account_type": account_type,"account_id": account_id,}params = {'constraints': json.dumps(constraints), 'page': 1, 'size': 3000}if self.tradingapiserver_host:url, headers = f"{self.tradingapiserver_host}/planned_order", Noneelse:url, headers = self.signature_headers(self.HOST, f"/bigapis/trading/v1/planned_order", self._access_key, self._secret_key, params=params)resp = requests.get(url=url, headers=headers, params=params)resp_data = resp.json()if resp_data["code"] != 0:raise Exception(resp_data["message"])list_planned_orders: List[Dict] = []######### For test mock planned orders"""_planned_order = {'planned_order_id': '1','strategy_id': self._strategy_id,'account_type': '0', 'account_id': '10000','trading_day': trading_day,'order_dt': ' '.join([trading_day, "09:31:00"]),'exchange': 'SH','instrument': '600900.SH','name': '长江电力','direction': '1','offset_flag': '0','original_order_qty': 600,'order_qty': 600,'order_price': 23.42,'order_type': 'U','order_status': 10,'status_msg': 'Generated'}list_planned_orders.append(_planned_order)return list_planned_orders"""########datas = resp_data.get("data").get("items")if datas:for planned_order in datas:if "created_at" in planned_order:del planned_order["created_at"]if "id" in planned_order:del planned_order["id"]if "updated_at" in planned_order:del planned_order["updated_at"]list_planned_orders.append(planned_order)else:list_planned_orders=list_planned_ordersif len(list_planned_orders):df=pd.DataFrame(list_planned_orders)df=df[['trading_day','instrument','name','original_order_qty','order_price','direction']]df.columns=['交易日','证券代码','名称','数量','价格','交易类型']df['交易类型']=df['交易类型'].apply(lambda x:'买' if x=='1' else '卖')else:df=pd.DataFrame()return dfdef signature_headers(self,host: str,path: str,access_key: str,secret_key: str,body: bytes = b"",headers: dict = {},params: dict = None,) -> Tuple[str, dict]:"""采用aksk给headers添加签名.Args:host (str): host.path (str): 路径.access_key (str): ak.secret_key (str): sk.body (bytes): 请求负载.headers (dict, optional): 请求头.Returns:Tuple[str, dict]: url, 请求头."""import hmacimport timefrom urllib.parse import quoteurl = f"{host}{path}"if params:url = url + "?" + quote("&".join([f"{k}={v}" for k, v in params.items()]))path_encode = path.encode()timestamp = int(time.time() * 1000)timestamp = str(timestamp)# 参与签名的消息msg = path_encode + body + timestamp.encode()signature = hmac.new(secret_key.encode(), msg=msg, digestmod="SHA256").hexdigest()# 给headers这是签名和时间戳headers[self.ACCESS_KEY_HEADER_NAME] = access_keyheaders[self.ACCESS_SIGNATURE_HEADER_NAME] = signatureheaders[self.ACCESS_TIMESTAMP_HEADER_NAME] = timestampreturn url, headers
if __name__ == "__main__":access_key = "J0kem11dvUy1"secret_key = "wnsvthHmHVuxHMStmNvGJyNxqR7lB0Eog7SfWDCkiIPiHfA3KwtCwoOebhqAUTiq"strategy_id = "4eeee2eb-fc9b-4b25-8442-d99c70d9ce5f"trading_day = '2025-05-23' # 假设交易日为当天# 初始化 API 对象api = xg_bigqunt_trader(access_key, secret_key)#先调这个函数获取参数账户类型#账户信息df=api.query_papertrading_info()print(df)#账户现金df=api.query_papertrading_cash()print(df)#持股df=api.query_papertrading_positions(strategy_id,trading_day)print(df)#持股全部df=api.query_papertrading_positions(strategy_id)print(df)#交易信号df=api.query_papertrading_planned_orders(strategy_id,trading_day)print(df)#全部交易信号df=api.query_papertrading_planned_orders_all(strategy_id)print(df)