IContext API 文档
由jliang创建，最终由jliang 被浏览 67 用户
IContext API 文档
IContext 接口类定义了 BigQuant AI 量化平台回测与交易引擎 bigtrader 策略 context 的抽象方法。StrategyContext 会继承 IContext，并实现具体的回测/实盘环境下的 context。 用户在编写策略时，handle_data 函数的第一个参数即为 IContext 类型对象，可以通过该对象访问平台提供的数据和交易接口。
1. 基本属性 (Basic Properties)
instruments
@property
def instruments(self) -> list[str]: ...
描述:
获取策略运行的标的列表。
返回值:
list[str]: 标的物代码列表，例如['000001.SZ', 'rb2010.SHF']。
示例:
def handle_data(context, data):
标的列表 = context.instruments
print(f"当前策略标的: {标的列表}")
data
@property
def data(self) -> Union["pd.DataFrame", Any]: ...
描述:
获取用户传入的数据。通常为 DataFrame 类型。
返回值:
Union[pd.DataFrame, Any]: 用户传入的数据，常见为 pandasDataFrame对象。
示例:
假设用户在 run 函数中传入了 data = pd.DataFrame({'close': [10, 11, 12]}, index=['20230101', '20230102', '20230103'])
import pandas as pd
def handle_data(context, data):
用户数据 = context.data
if isinstance(用户数据, pd.DataFrame):
print(f"用户数据 (前两行):\n{用户数据.head(2)}")
options
@property
def options(self) -> dict[str, Any]: ...
描述:
获取扩展数据。可以通过 options['key'] 的方式访问。
返回值:
dict[str, Any]: 包含扩展数据的字典。
示例:
假设用户在 run 函数中传入了 options = {'factor_data': factor_df, 'strategy_params': {'param1': 100}}
def handle_data(context, data):
扩展选项 = context.options
因子数据 = 扩展选项.get('factor_data')
策略参数 = 扩展选项.get('strategy_params')
if 因子数据 is not None:
print(f"因子数据类型: {type(因子数据)}")
if 策略参数 is not None:
print(f"策略参数: {策略参数}")
user_store
@property
def user_store(self) -> dict[str, Any]: ...
描述:
获取用户全局变量存储字典。用于存放用户的全局变量。在模拟交易等模式下会做持久化存储，自动保存和恢复状态。注意：存储的数据类型需要支持 JSON 序列化。
返回值:
dict[str, Any]: 用户全局变量存储字典。
示例:
def handle_data(context, data):
全局变量存储 = context.user_store
计数器 = 全局变量存储.get('my_counter', 0)
计数器 += 1
全局变量存储['my_counter'] = 计数器
print(f"计数器当前值: {计数器}")
run_mode
@property
def run_mode(self): ...
描述:
获取当前运行模式。
返回值:
RunMode: 当前的运行模式，可能的值为RunMode.BACKTEST(回测) 或RunMode.LIVE(实盘/模拟交易)。 (需要导入RunMode枚举类型)
示例:
def handle_data(context, data):
运行模式 = context.run_mode
if 运行模式 == RunMode.BACKTEST:
print("当前处于回测模式")
elif 运行模式 == RunMode.LIVE:
print("当前处于实盘/模拟交易模式")
2. 交易日历 (Trading Calendar)
trading_calendar
@property
def trading_calendar(self): ...
描述:
获取交易日历对象。
返回值:
TradingCalendar: 交易日历对象。 (需要导入TradingCalendar类型)
first_trading_date
@property
def first_trading_date(self): ...
描述:
获取回测或实盘交易周期的第一个交易日。
返回值:
str: 第一个交易日，格式为YYYYmmdd。
示例:
def handle_data(context, data):
首个交易日 = context.first_trading_date
print(f"首个交易日: {首个交易日}")
trading_day_index
@property
def trading_day_index(self): ...
描述:
获取当前交易日在交易日历中的索引位置。
返回值:
int: 当前交易日索引。
示例:
def handle_data(context, data):
交易日索引 = context.trading_day_index
print(f"当前交易日索引: {交易日索引}")
get_trading_day
@abstractmethod
def get_trading_day(self) -> str: ...
描述:
获取当前交易日。
返回值:
str: 当前交易日，格式为YYYYmmdd。
示例:
def handle_data(context, data):
当前交易日 = context.get_trading_day()
print(f"当前交易日: {当前交易日}")
get_prev_trading_day
@abstractmethod
def get_prev_trading_day(self, cur_trading_day: str = "") -> str: ...
描述:
获取指定交易日的前一个交易日。
参数:
cur_trading_day(str, 可选): 当前交易日，格式为YYYYmmdd。如果为空字符串，则默认为当前交易日。
返回值:
str: 前一个交易日，格式为YYYYmmdd。
示例:
def handle_data(context, data):
当前交易日 = context.get_trading_day()
前一个交易日 = context.get_prev_trading_day(cur_trading_day=当前交易日)
print(f"当前交易日: {当前交易日}, 前一个交易日: {前一个交易日}")
get_next_trading_day
@abstractmethod
def get_next_trading_day(self, cur_trading_day: str = "") -> str: ...
描述:
获取指定交易日的后一个交易日。
参数:
cur_trading_day(str, 可选): 当前交易日，格式为YYYYmmdd。如果为空字符串，则默认为当前交易日。
返回值:
str: 后一个交易日，格式为YYYYmmdd。
示例:
def handle_data(context, data):
当前交易日 = context.get_trading_day()
后一个交易日 = context.get_next_trading_day(cur_trading_day=当前交易日)
print(f"当前交易日: {当前交易日}, 后一个交易日: {后一个交易日}")
have_night_trading
@abstractmethod
def have_night_trading(self, trading_day: str) -> bool: ...
描述:
检查指定交易日是否有夜盘交易。
参数:
trading_day(str): 交易日，格式为YYYYmmdd。
返回值:
bool:True如果该交易日有夜盘交易，False则没有。
示例:
def handle_data(context, data):
当前交易日 = context.get_trading_day()
是否有夜盘 = context.have_night_trading(当前交易日)
if 是否有夜盘:
print(f"{当前交易日} 有夜盘交易")
else:
print(f"{当前交易日} 没有夜盘交易")
3. 账户管理 (Account Management)
portfolio
@property
def portfolio(self): ...
描述:
获取策略账户的投资组合对象。通过该对象可以访问账户余额、持仓等信息。
返回值:
Portfolio: 投资组合对象。 (需要导入Portfolio类型)
add_account
@abstractmethod
def add_account(self, acct_type: AcctType, account_id: str = "", capital_base: float = 1e06, **kwargs): ...
描述:
[回测专用] 初始化额外的回测交易账户。用于在回测中创建和管理多个交易账户，例如股票账户、期货账户、期权账户等。
参数:
acct_type(AcctType): 账户类型，例如AcctType.STOCK(股票账户),AcctType.FUTURE(期货账户),AcctType.OPTION(期权账户)。 (需要导入AcctType枚举类型)account_id(str, 可选): 自定义账户 ID。如果不提供，则自动生成默认 ID。默认为""。capital_base(float, 可选): 回测账户的初始资金。默认为1e06(一百万)。**kwargs: 额外的关键字参数，用于账户初始化设置。
示例 (回测中添加一个期货账户):
def initialize(context):
context.add_account(acct_type=AcctType.FUTURE, account_id="future_account", capital_base=500000)
def handle_data(context, data):
# ... 在主账户 (默认账户) 上进行股票交易 ...
# ... 在 "future_account" 期货账户上进行期货交易 ...
pass
login_account
def login_account(self, account_setting: dict): ...
描述:
[实盘/模拟交易专用] 尝试登录其他交易账户。用于在实盘或模拟交易中连接到券商账户。
参数:
account_setting(dict): 账户登录设置字典。具体需要提供的键值对取决于不同的券商 API。
示例 (模拟账户登录，假设 account_setting 包含了必要的登录信息):
def initialize(context):
account_config = {
'account_id': 'your_sim_account_id',
'password': 'your_password',
'broker': 'example_broker'
}
context.login_account(account_setting=account_config)
def handle_data(context, data):
# ... 进行实盘/模拟交易 ...
pass
get_balance
@abstractmethod
def get_balance(self, account_id: str = "") -> float: ...
描述:
获取交易账户的总资金余额。
参数:
account_id(str, 可选): 账户 ID。如果不指定，则使用默认账户。默认为""。
返回值:
float: 账户总资金余额。
示例:
def handle_data(context, data):
账户余额 = context.get_balance() # 获取默认账户余额
print(f"默认账户余额: {账户余额}")
期货账户余额 = context.get_balance(account_id="future_account") # 获取指定期货账户余额 (如果已添加)
if 期货账户余额 is not None:
print(f"期货账户余额: {期货账户余额}")
get_available_cash
@abstractmethod
def get_available_cash(self, account_id: str = "") -> float: ...
描述:
获取交易账户的可用资金余额。即可用于下单的现金。
参数:
account_id(str, 可选): 账户 ID。如果不指定，则使用默认账户。默认为""。
返回值:
float: 账户可用资金余额。
示例:
def handle_data(context, data):
可用资金 = context.get_available_cash() # 获取默认账户可用资金
print(f"默认账户可用资金: {可用资金}")
期货账户可用资金 = context.get_available_cash(account_id="future_account") # 获取指定期货账户可用资金 (如果已添加)
if 期货账户可用资金 is not None:
print(f"期货账户可用资金: {期货账户可用资金}")
4. 持仓管理 (Position Management)
get_position
@abstractmethod
def get_position(self, symbol: str, direction: Direction = Direction.NONE, create_if_none: bool = True, account_id: str = "") -> Any: ...
描述:
获取指定标的、指定方向的持仓对象。
参数:
symbol(str): 标的物代码，例如'000001.SZ','rb2010.SHF'。direction(Direction, 可选): 持仓方向。Direction.LONG(多仓),Direction.SHORT(空仓),Direction.NONE(净持仓)。默认为Direction.NONE。 (需要导入Direction枚举类型)create_if_none(bool, 可选): 如果指定标的和方向的持仓不存在，是否创建新的持仓对象。默认为True。account_id(str, 可选): 账户 ID。如果不指定，则使用默认账户。默认为""。
返回值:
Position: 持仓对象。如果持仓不存在且create_if_none为False，则返回None。 (需要导入Position类型)
示例:
def handle_data(context, data):
股票持仓 = context.get_position(symbol='000001.SZ') # 获取默认账户 000001.SZ 的净持仓
if 股票持仓:
print(f"000001.SZ 持仓量: {股票持仓.volume}")
期货多仓 = context.get_position(symbol='rb2010.SHF', direction=Direction.LONG, account_id="future_account") # 获取指定期货账户 rb2010.SHF 的多仓
if 期货多仓:
print(f"rb2010.SHF 多仓持仓量 (期货账户): {期货多仓.volume}")
get_positions
@abstractmethod
def get_positions(self, symbols: Union[List[str], None] = None, account_id: str = "") -> Dict[str, Any]: ...
描述:
获取账户的所有持仓，或指定标的列表的持仓。
参数:
symbols(Union[List[str], None], 可选): 标的物代码列表。如果为None，则返回账户所有标的的持仓。默认为None。account_id(str, 可选): 账户 ID。如果不指定，则使用默认账户。默认为""。
返回值:
Dict[str, Position]: 一个字典，键为标的物代码，值为Position持仓对象。
示例:
def handle_data(context, data):
所有持仓 = context.get_positions() # 获取默认账户所有持仓
if 所有持仓:
print("默认账户所有持仓:")
for symbol, position in 所有持仓.items():
print(f" {symbol}: 持仓量 {position.volume}")
指定标的持仓 = context.get_positions(symbols=['000001.SZ', '600519.SH']) # 获取默认账户指定标的的持仓
if 指定标的持仓:
print("指定标的持仓:")
for symbol, position in 指定标的持仓.items():
print(f" {symbol}: 持仓量 {position.volume}")
set_initial_positions
@abstractmethod
def set_initial_positions(self, initial_positions: list, account_id: str = ""): ...
描述:
[回测专用] 批量设置账户的初始持仓。用于在回测开始时预设账户的持仓情况。
参数:
initial_positions(list): 初始持仓列表。列表中的每个元素应为持仓对象或描述持仓信息的字典。具体格式取决于Position对象的结构。account_id(str, 可选): 账户 ID。如果不指定，则使用默认账户。默认为""。
示例 (回测中设置初始持仓):
def initialize(context):
initial_positions_data = [
{'symbol': '000001.SZ', 'volume': 1000}, # 假设字典格式可以接受
{'symbol': '600519.SH', 'volume': 500}
]
context.set_initial_positions(initial_positions=initial_positions_data)
def handle_data(context, data):
# ... 回测逻辑 ...
pass
5. 订单操作 (Order Placement)
order
@abstractmethod
def order(self, symbol: str, volume: int, limit_price: Union[float, None] = None, offset: Union[Offset, None] = None, order_type: Union[OrderType, None] = None, **kwargs) -> int: ...
描述:
下单接口。这是最通用的下单函数，允许设置各种订单参数。
参数:
symbol(str): 标的物代码，例如'000001.SZ','rb2010.SHF'。volume(int): 订单数量。正数表示买入，负数表示卖出。limit_price(Union[float, None], 可选): 限价单价格。如果为None，并且order_type为OrderType.MARKET，则为市价单。 默认为None。offset(Union[Offset, None], 可选): 期货/期权开平仓类型。Offset.OPEN(开仓),Offset.CLOSE(平仓),Offset.CLOSETODAY(平今仓)。如果为None，系统自动判断开平仓。 默认为None。 (需要导入Offset枚举类型)order_type(Union[OrderType, None], 可选): 订单类型。OrderType.LIMIT(限价单),OrderType.MARKET(市价单)。如果为None，默认为OrderType.LIMIT。 (需要导入OrderType枚举类型)**kwargs: 可选的关键字参数，例如order_property(订单属性),hedge_flag(套保标志) 等，具体参数取决于交易所或券商 API 的支持。
返回值:
int:0表示下单成功，其他非零值表示错误代码。可以使用get_error_msg(error_id)获取详细错误信息。
示例 (限价买入 100 股 000001.SZ，价格 10.5 元):
def handle_data(context, data):
order_id = context.order(symbol='000001.SZ', volume=100, limit_price=10.5)
if order_id == 0:
print("下单成功")
else:
error_msg = context.get_error_msg(order_id)
print(f"下单失败，错误信息: {error_msg}")
示例 (市价卖出 5 手 rb2010.SHF，平仓):
def handle_data(context, data):
order_id = context.order(symbol='rb2010.SHF', volume=-5, offset=Offset.CLOSE, order_type=OrderType.MARKET)
if order_id == 0:
print("下单成功")
else:
error_msg = context.get_error_msg(order_id)
print(f"下单失败，错误信息: {error_msg}")
order_percent
@abstractmethod
def order_percent(self, symbol: str, percent: float, limit_price: Union[float, None] = None, order_type: Union[OrderType, None] = None, **kwargs) -> int: ...
描述:
按可用资金的百分比下单。
参数:
symbol(str): 标的物代码。percent(float): 下单金额占可用资金的百分比，例如0.1表示 10%。limit_price(Union[float, None], 可选): 限价单价格。默认为None。order_type(Union[OrderType, None], 可选): 订单类型。默认为None。**kwargs: 可选的关键字参数。
返回值:
int:0表示下单成功，其他非零值表示错误代码。
示例 (买入价值 10% 可用资金的 000001.SZ，限价单):
def handle_data(context, data):
order_id = context.order_percent(symbol='000001.SZ', percent=0.1, limit_price=10.6)
if order_id == 0:
print("下单成功")
else:
error_msg = context.get_error_msg(order_id)
print(f"下单失败，错误信息: {error_msg}")
order_value
@abstractmethod
def order_value(self, symbol: str, value: float, limit_price: Union[float, None] = None, order_type: Union[OrderType, None] = None, **kwargs) -> int: ...
描述:
按目标价值下单。
参数:
symbol(str): 标的物代码。value(float): 下单的目标价值，单位为账户的计价货币，例如人民币10000表示下单价值 10000 元人民币。limit_price(Union[float, None], 可选): 限价单价格。默认为None。order_type(Union[OrderType, None], 可选): 订单类型。默认为None。**kwargs: 可选的关键字参数。
返回值:
int:0表示下单成功，其他非零值表示错误代码。
示例 (买入价值 10000 元的 000001.SZ，市价单):
def handle_data(context, data):
order_id = context.order_value(symbol='000001.SZ', value=10000, order_type=OrderType.MARKET)
if order_id == 0:
print("下单成功")
else:
error_msg = context.get_error_msg(order_id)
print(f"下单失败，错误信息: {error_msg}")
order_target
@abstractmethod
def order_target(self, symbol: str, target: float, limit_price: Union[float, None] = None, order_type: Union[OrderType, None] = None, **kwargs) -> int: ...
描述:
调整仓位到目标数量。计算达到目标持仓量所需的交易数量并下单。
参数:
symbol(str): 标的物代码。target(float): 目标持仓数量。limit_price(Union[float, None], 可选): 限价单价格。默认为None。order_type(Union[OrderType, None], 可选): 订单类型。默认为None。**kwargs: 可选的关键字参数。
返回值:
int:0表示下单成功，其他非零值表示错误代码。
示例 (调整 000001.SZ 持仓到目标 1000 股，限价单):
def handle_data(context, data):
order_id = context.order_target(symbol='000001.SZ', target=1000, limit_price=10.7)
if order_id == 0:
print("下单成功")
else:
error_msg = context.get_error_msg(order_id)
print(f"下单失败，错误信息: {error_msg}")
order_target_percent
@abstractmethod
def order_target_percent(self, symbol: str, target: float, limit_price: Union[float, None] = None, order_type: Union[OrderType, None] = None, **kwargs) -> int: ...
描述:
调整仓位到目标百分比。计算达到目标持仓百分比所需的交易数量并下单。
参数:
symbol(str): 标的物代码。target(float): 目标持仓占总资产的百分比，例如0.2表示目标持仓价值占总资产的 20%。limit_price(Union[float, None], 可选): 限价单价格。默认为None。order_type(Union[OrderType, None], 可选): 订单类型。默认为None。**kwargs: 可选的关键字参数。
返回值:
int:0表示下单成功，其他非零值表示错误代码。
示例 (调整 000001.SZ 持仓到目标 20% 总资产，市价单):
def handle_data(context, data):
order_id = context.order_target_percent(symbol='000001.SZ', target=0.2, order_type=OrderType.MARKET)
if order_id == 0:
print("下单成功")
else:
error_msg = context.get_error_msg(order_id)
print(f"下单失败，错误信息: {error_msg}")
order_target_value
@abstractmethod
def order_target_value(self, symbol: str, target: float, limit_price: Union[float, None] = None, order_type: Union[OrderType, None] = None, **kwargs) -> int: ...
描述:
调整仓位到目标价值。计算达到目标持仓价值所需的交易数量并下单。
参数:
symbol(str): 标的物代码。target(float): 目标持仓价值，单位为账户的计价货币。limit_price(Union[float, None], 可选): 限价单价格。默认为None。order_type(Union[OrderType, None], 可选): 订单类型。默认为None。**kwargs: 可选的关键字参数。
返回值:
int:0表示下单成功，其他非零值表示错误代码。
示例 (调整 000001.SZ 持仓到目标价值 20000 元，限价单):
def handle_data(context, data):
order_id = context.order_target_value(symbol='000001.SZ', target=20000, limit_price=10.8)
if order_id == 0:
print("下单成功")
else:
error_msg = context.get_error_msg(order_id)
print(f"下单失败，错误信息: {error_msg}")
place_order
@abstractmethod
def place_order(self, req: OrderReq, **kwargs) -> int: ...
描述:
使用 OrderReq 对象下单。允许用户自定义更详细的订单请求参数。
参数:
req(OrderReq):OrderReq订单请求对象，包含订单的详细信息。 (需要导入OrderReq类型)**kwargs: 可选的关键字参数。
返回值:
int:0表示下单成功，其他非零值表示错误代码。
示例 (使用 OrderReq 对象下单):
from bigtrader.request import OrderReq # 假设 OrderReq 定义在这个模块
def handle_data(context, data):
order_req = OrderReq(
symbol='000001.SZ',
volume=200,
limit_price=10.9,
order_type=OrderType.LIMIT
)
order_id = context.place_order(req=order_req)
if order_id == 0:
print("下单成功")
else:
error_msg = context.get_error_msg(order_id)
print(f"下单失败，错误信息: {error_msg}")
buy_open
@abstractmethod
def buy_open(self, symbol: str, volume: int, limit_price: Union[float, None] = None, order_type: Union[OrderType, None] = None, **kwargs) -> int: ...
描述:
买入开仓下单。明确指定为买入开仓操作 (对于期货/期权)。
参数:
symbol(str): 标的物代码。volume(int): 买入数量 (正数)。limit_price(Union[float, None], 可选): 限价单价格。默认为None。order_type(Union[OrderType, None], 可选): 订单类型。默认为None。**kwargs: 可选的关键字参数。
返回值:
int:0表示下单成功，其他非零值表示错误代码。
buy_close
@abstractmethod
def buy_close(self, symbol: str, volume: int, limit_price: Union[float, None] = None, order_type: Union[OrderType, None] = None, **kwargs) -> int: ...
描述:
买入平仓下单。明确指定为买入平仓操作 (对于期货/期权)。
参数:
symbol(str): 标的物代码。volume(int): 买入数量 (正数)。limit_price(Union[float, None], 可选): 限价单价格。默认为None。order_type(Union[OrderType, None], 可选): 订单类型。默认为None。**kwargs: 可选的关键字参数。
返回值:
int:0表示下单成功，其他非零值表示错误代码。
sell_open
@abstractmethod
def sell_open(self, symbol: str, volume: int, limit_price: Union[float, None] = None, order_type: Union[OrderType, None] = None, **kwargs) -> int: ...
描述:
卖出开仓下单。明确指定为卖出开仓操作 (对于期货/期权)。
参数:
symbol(str): 标的物代码。volume(int): 卖出数量 (正数)。limit_price(Union[float, None], 可选): 限价单价格。默认为None。order_type(Union[OrderType, None], 可选): 订单类型。默认为None。**kwargs: 可选的关键字参数。
返回值:
int:0表示下单成功，其他非零值表示错误代码。
sell_close
@abstractmethod
def sell_close(self, symbol: str, volume: int, limit_price: Union[float, None] = None, order_type: Union[OrderType, None] = None, **kwargs) -> int: ...
描述:
卖出平仓下单。明确指定为卖出平仓操作 (对于期货/期权)。
参数:
symbol(str): 标的物代码。volume(int): 卖出数量 (正数)。limit_price(Union[float, None], 可选): 限价单价格。默认为None。order_type(Union[OrderType, None], 可选): 订单类型。默认为None。**kwargs: 可选的关键字参数。
返回值:
int:0表示下单成功，其他非零值表示错误代码。
exercise
@abstractmethod
def exercise(self, symbol: str, volume: int, **kwargs): ...
描述:
期权行权下单。
参数:
symbol(str): 期权合约代码。volume(int): 行权数量。**kwargs: 可选的关键字参数。
示例 (行权 10 张 510050.SH 认购期权):
def handle_data(context, data):
context.exercise(symbol='某期权合约代码', volume=10)
print("期权行权下单已提交")
order_reverse_repo
@abstractmethod
def order_reverse_repo(self, value_or_percent: float, n: int = 1, rate: Union[float, None] = None, **kwargs): ...
描述:
逆回购下单。借出现金以获得利息。
参数:
value_or_percent(float): 逆回购金额。如果值<= 1.0，则表示使用可用资金的百分比，否则表示具体金额。n(int, 可选): 逆回购天数。默认为1天。rate(Union[float, None], 可选): 交易利率，默认年化利率为1.1%(具体数值可能会调整)。默认为None，使用平台默认利率。**kwargs: 可选的关键字参数。
示例 (使用 50% 可用资金进行 1 天逆回购):
def handle_data(context, data):
context.order_reverse_repo(value_or_percent=0.5, n=1)
print("逆回购下单已提交 (50% 可用资金，1天)")
示例 (使用 100000 元进行 7 天逆回购，指定年化利率为 2.0%):
def handle_data(context, data):
context.order_reverse_repo(value_or_percent=100000, n=7, rate=0.02) # 假设 rate 0.02 表示 2% 年化利率
print("逆回购下单已提交 (100000元，7天，年化利率 2.0%)")
6. 订单查询与撤销 (Order Query and Cancellation)
cancel_order
@abstractmethod
def cancel_order(self, order_param: Union[str, OrderData, OrderReq]) -> int: ...
描述:
撤销指定订单。
参数:
order_param(Union[str, OrderData, OrderReq]): 订单标识。可以是order_key(字符串),OrderData对象, 或CancelOrderReq对象。 (需要导入OrderData,CancelOrderReq类型)
返回值:
int:0表示撤单成功，其他非零值表示错误代码。
示例 (根据 order_key 撤单):
def handle_data(context, data):
last_order_key = context.get_last_order_key() # 获取最近一次下单成功的 order_key
if last_order_key:
cancel_result = context.cancel_order(order_param=last_order_key)
if cancel_result == 0:
print(f"订单 {last_order_key} 撤单成功")
else:
error_msg = context.get_error_msg(cancel_result)
print(f"订单 {last_order_key} 撤单失败，错误信息: {error_msg}")
else:
print("没有最近下单的订单 key")
cancel_all
@abstractmethod
def cancel_all(self, only_from_this: bool = False, account_id: str = "") -> None: ...
描述:
撤销所有未成交的订单。
参数:
only_from_this(bool, 可选): 如果为True，则只撤销当前策略实例的订单。默认为False，撤销账户下的所有订单。account_id(str, 可选): 账户 ID。如果不指定，则使用默认账户。默认为""。
示例 (撤销当前策略实例的所有未成交订单):
def handle_data(context, data):
context.cancel_all(only_from_this=True)
print("已尝试撤销当前策略的所有未成交订单")
get_open_orders
@abstractmethod
def get_open_orders(self, symbol: str, only_from_this: bool = False) -> List[Union[OrderData, OrderReq]]: ...
描述:
获取指定标的的未成交订单列表。
参数:
symbol(str): 标的物代码。only_from_this(bool, 可选): 如果为True，则只获取当前策略实例的订单。默认为False，获取所有来源的订单。
返回值:
List[Union[OrderData, OrderReq]]: 未成交订单列表。列表中的元素可能是OrderData对象或OrderReq对象。
示例 (获取 000001.SZ 的未成交订单):
def handle_data(context, data):
open_orders = context.get_open_orders(symbol='000001.SZ')
if open_orders:
print(f"000001.SZ 未成交订单:")
for order in open_orders:
print(f" 订单号: {order.order_id}, 状态: {order.status}") # 假设 OrderData/OrderReq 有 order_id 和 status 属性
else:
print(f"000001.SZ 没有未成交订单")
get_orders
@abstractmethod
def get_orders(self, symbol: str, only_from_this: bool = False) -> List[OrderData]: ...
描述:
获取指定标的的所有订单列表 (包括已成交和未成交)。
参数:
symbol(str): 标的物代码。only_from_this(bool, 可选): 如果为True，则只获取当前策略实例的订单。默认为False，获取所有来源的订单。
返回值:
List[OrderData]: 所有订单列表，元素为OrderData对象。
示例 (获取 000001.SZ 的所有订单):
def handle_data(context, data):
all_orders = context.get_orders(symbol='000001.SZ')
if all_orders:
print(f"000001.SZ 所有订单:")
for order in all_orders:
print(f" 订单号: {order.order_id}, 状态: {order.status}")
else:
print(f"000001.SZ 没有订单记录")
get_order
@abstractmethod
def get_order(self, order_param: str, account_id: str = "") -> Any: ...
描述:
获取指定订单的详细信息。
参数:
order_param(str): 订单标识，可以是order_key或bt_order_sysid。account_id(str, 可选): 账户 ID。如果不指定，则使用默认账户。默认为""。
返回值:
OrderData: 订单对象。如果未找到订单，可能返回None。
示例 (根据 order_key 获取订单详情):
def handle_data(context, data):
last_order_key = context.get_last_order_key()
if last_order_key:
order_detail = context.get_order(order_param=last_order_key)
if order_detail:
print(f"订单 {last_order_key} 详情: 订单状态: {order_detail.status}, 成交均价: {order_detail.avg_price}") # 假设 OrderData 有 status 和 avg_price 属性
else:
print(f"未找到订单 {last_order_key} 的详细信息")
else:
print("没有最近下单的订单 key")
get_trades
@abstractmethod
def get_trades(self, symbol: str = "", only_from_this: bool = False) -> List[TradeData]: ...
描述:
获取指定标的或所有标的的成交记录。
参数:
symbol(str, 可选): 标的物代码。如果为空字符串，则获取所有标的的成交记录。默认为""。only_from_this(bool, 可选): 如果为True，则只获取当前策略实例的成交记录。默认为False，获取所有来源的成交记录。
返回值:
List[TradeData]: 成交记录列表，元素为TradeData对象。 (需要导入TradeData类型)
示例 (获取 000001.SZ 的成交记录):
def handle_data(context, data):
trades = context.get_trades(symbol='000001.SZ')
if trades:
print(f"000001.SZ 成交记录:")
for trade in trades:
print(f" 成交时间: {trade.trade_time}, 成交价格: {trade.price}, 成交数量: {trade.volume}") # 假设 TradeData 有 trade_time, price, volume 属性
else:
print(f"000001.SZ 没有成交记录")
get_pending_order_reqs
@abstractmethod
def get_pending_order_reqs(self, account_id: str = "") -> List[OrderReq]: ...
描述:
获取当日待处理的订单请求列表。待处理订单请求指已提交但尚未完全被系统处理的订单。
参数:
account_id(str, 可选): 账户 ID。如果不指定，则使用默认账户。默认为""。
返回值:
List[OrderReq]: 待处理订单请求列表，元素为OrderReq对象。
remove_pending_order_req
@abstractmethod
def remove_pending_order_req(self, order_req: OrderReq): ...
描述:
移除当日待处理的订单请求。
参数:
order_req(OrderReq): 要移除的OrderReq订单请求对象。
get_last_order_key
@abstractmethod
def get_last_order_key(self) -> str: ...
描述:
获取最近一次下单成功的 order_key。
返回值:
str: 最近一次下单成功的order_key字符串。如果还没有成功下单过，则返回None。
7. 数据订阅 (Data Subscription)
subscribe
@abstractmethod
def subscribe(self, symbols: Union[str, List[str]], subscribe_flags: SubscribeFlag = SubscribeFlag.DEFAULT): ...
描述:
订阅市场数据。
参数:
symbols(Union[str, List[str]]): 要订阅的标的物代码，可以是单个代码字符串或代码列表。subscribe_flags(SubscribeFlag, 可选): 订阅标志，用于指定订阅的数据类型。默认为SubscribeFlag.DEFAULT，根据频率设置订阅默认数据。 (需要导入SubscribeFlag枚举类型)SubscribeFlag.DEFAULT (0): 默认订阅，根据频率设置决定订阅的数据类型。SubscribeFlag.LEVEL1_SNAPSHOT (1): Level1 快照行情 (对应handle_tick函数)。SubscribeFlag.LEVEL2_SNAPSHOT (2): Level2 快照行情 (对应handle_tick函数)。SubscribeFlag.LEVEL2_TRADE (4): Level2 逐笔成交 (对应handle_l2trade函数)。SubscribeFlag.LEVEL2_ORDER (8): Level2 委托队列 (对应handle_l2order函数)。- 可以使用 位或运算
|组合多个标志，例如SubscribeFlag.LEVEL2_SNAPSHOT | SubscribeFlag.LEVEL2_TRADE表示同时订阅 Level2 快照和逐笔成交。
示例 (订阅 000001.SZ 和 600519.SH 的默认行情):
def initialize(context):
context.subscribe(symbols=['000001.SZ', '600519.SH'])
print("已订阅 000001.SZ 和 600519.SH 的默认行情")
def handle_data(context, data):
# ... 在 handle_data 中处理行情数据 ...
pass
示例 (订阅 rb2010.SHF 的 Level2 快照和逐笔成交行情):
def initialize(context):
context.subscribe(symbols='rb2010.SHF', subscribe_flags=SubscribeFlag.LEVEL2_SNAPSHOT | SubscribeFlag.LEVEL2_TRADE)
print("已订阅 rb2010.SHF 的 Level2 快照和逐笔成交行情")
def handle_tick(context, tick_data): # 处理快照行情
print(f"收到 Tick 快照数据: {tick_data}")
def handle_l2trade(context, l2trade_data): # 处理逐笔成交行情
print(f"收到 Level2 逐笔成交数据: {l2trade_data}")
unsubscribe
@abstractmethod
def unsubscribe(self, symbols: Union[str, List[str]], subscribe_flags: SubscribeFlag = SubscribeFlag.DEFAULT): ...
描述:
取消订阅市场数据。
参数:
symbols(Union[str, List[str]]): 要取消订阅的标的物代码，可以是单个代码字符串或代码列表。subscribe_flags(SubscribeFlag, 可选): 取消订阅的标志，应与订阅时使用的标志一致。默认为SubscribeFlag.DEFAULT。
示例 (取消订阅 000001.SZ 的默认行情):
def initialize(context):
context.unsubscribe(symbols='000001.SZ')
print("已取消订阅 000001.SZ 的默认行情")
subscribe_bar
@abstractmethod
def subscribe_bar(self, symbols: Union[List[str], str], period: str, callback: Union[Callable, None] = None, history_window: Union[int, None] = None): ...
描述:
订阅 Bar (K线) 行情数据。
参数:
symbols(Union[List[str], str]): 要订阅 Bar 行情的标的物代码，可以是单个代码字符串或代码列表。period(str): Bar 周期，例如'1m'(1分钟),'5m'(5分钟),'1d'(日线) 等。callback(Union[Callable, None], 可选): Bar 数据回调函数。当收到新的 Bar 数据时，会调用此函数。函数应接受两个参数:context(策略 context 对象) 和bar(Bar 数据对象)。例如handle_bar(context, bar)。如果为None，则默认使用handle_bar函数。history_window(Union[int, None], 可选): 历史 Bar 数据窗口大小。用于策略中需要使用data.history()等方法访问历史数据的场景。设置后，平台会维护指定长度的历史 Bar 数据。默认为None，不维护历史数据窗口。
示例 (订阅 000001.SZ 的 5 分钟 Bar 行情):
def initialize(context):
context.subscribe_bar(symbols='000001.SZ', period='5m')
print("已订阅 000001.SZ 的 5 分钟 Bar 行情")
def handle_bar(context, bar_data): # 处理 5 分钟 Bar 行情
print(f"收到 5 分钟 Bar 数据: {bar_data}")
示例 (订阅 600519.SH 的 1 日 Bar 行情，并设置历史数据窗口为 100 天):
def initialize(context):
context.subscribe_bar(symbols='600519.SH', period='1d', history_window=100)
print("已订阅 600519.SH 的 1 日 Bar 行情，历史窗口 100 天")
def handle_bar(context, bar_data):
# ... 可以使用 data.history('600519.SH', 'close', '1d', 10) 获取最近 10 天的日线收盘价 ...
pass
8. 数据查询 (Data Retrieval)
get_contract
@abstractmethod
def get_contract(self, symbol: str) -> ContractData: ...
描述:
获取标的物合约信息。
参数:
symbol(str): 标的物代码，例如'000001.SZ','rb2010.SHF'。
返回值:
ContractData: 合约信息对象。如果未找到合约信息，则返回None。 (需要导入ContractData类型)
示例 (获取 000001.SZ 的合约信息):
def handle_data(context, data):
contract_info = context.get_contract(symbol='000001.SZ')
if contract_info:
print(f"000001.SZ 合约名称: {contract_info.name}, 交易单位: {contract_info.multiplier}") # 假设 ContractData 有 name 和 multiplier 属性
else:
print(f"未找到 000001.SZ 的合约信息")
get_all_contracts
@abstractmethod
def get_all_contracts(self, product_code: str = "", **kwargs) -> Dict[str, ContractData]: ...
描述:
获取所有合约信息，或根据产品代码过滤合约。
参数:
product_code(str, 可选): 产品代码，例如'rb'(螺纹钢期货)。如果为空字符串，则返回所有合约信息。默认为""。**kwargs: 额外的关键字参数，用于过滤或选项设置。
返回值:
Dict[str, ContractData]: 合约信息字典，键为标的物代码，值为ContractData对象。
示例 (获取所有螺纹钢期货合约信息):
def handle_data(context, data):
rebar_contracts = context.get_all_contracts(product_code='rb')
if rebar_contracts:
print("螺纹钢期货合约列表:")
for symbol, contract in rebar_contracts.items():
print(f" {symbol}: 合约名称 {contract.name}")
else:
print("未找到螺纹钢期货合约信息")
get_all_option_contracts
@abstractmethod
def get_all_option_contracts(self, underlying: str) -> Dict[str, ContractData]: ...
描述:
查询某标的的所有期权合约。
参数:
underlying(str): 标的物代码，例如'a2309.DCE'(期货期权标的),'000016.SH'(指数期权标的),'510050.SH'(ETF 期权标的)。
返回值:
Dict[str, OptionContractData]: 期权合约信息字典，键为期权合约代码，值为OptionContractData对象。 (需要导入OptionContractData类型，这里文档假设返回的是ContractData，需要确认实际类型)
示例 (获取 510050.SH (ETF) 的所有期权合约):
def handle_data(context, data):
option_contracts = context.get_all_option_contracts(underlying='510050.SH')
if option_contracts:
print("510050.SH 期权合约列表:")
for symbol, contract in option_contracts.items():
print(f" {symbol}: 合约名称 {contract.name}, 期权类型: {contract.option_type}, 行权价: {contract.strike}") # 假设 OptionContractData 有 option_type 和 strike 属性
else:
print("未找到 510050.SH 的期权合约信息")
get_option_strike_prices
@abstractmethod
def get_option_strike_prices(self, underlying: str, strike_year_month: int) -> List[float]: ...
描述:
获取某标的下指定年月的所有期权执行价格。
参数:
underlying(str): 标的物代码，例如'a2309.DCE','000016.SH','510050.SH'。strike_year_month(int): 执行价格的年月，格式为YYYYmm，例如202009表示 2020 年 9 月。
返回值:
List[float]: 执行价格列表。
示例 (获取 510050.SH 在 2023 年 12 月到期的所有期权执行价格):
def handle_data(context, data):
strike_prices = context.get_option_strike_prices(underlying='510050.SH', strike_year_month=202312)
if strike_prices:
print(f"510050.SH 在 202312 到期的期权执行价格:")
for price in strike_prices:
print(f" {price}")
else:
print(f"未找到 510050.SH 在 202312 到期的期权执行价格")
get_atm_option_contract
@abstractmethod
def get_atm_option_contract(self, underlying: str, strike_year_month: int, underlying_price: float, option_type: OptionCP, adjust_flag: str = "A") -> ContractData: ...
描述:
获取某档位期权合约。通常用于获取平值期权合约。
参数:
underlying(str): 标的物代码，例如'a2309.DCE','000016.SH','510050.SH'。strike_year_month(int): 执行价格的年月，格式为YYYYmm，例如'202009'。underlying_price(float): 标的物当前价格。option_type(OptionCP): 期权类型，OptionCP.CALL(认购期权) 或OptionCP.PUT(认沽期权)。 (需要导入OptionCP枚举类型)adjust_flag(str, 可选): 调整标志。默认为"A"，具体含义可能与数据提供商有关。
返回值:
OptionContractData: 期权合约对象。如果未找到，则返回None。 (需要导入OptionContractData类型，这里文档假设返回的是ContractData，需要确认实际类型)
示例 (获取 510050.SH 在 2023 年 12 月到期的平值认购期权):
def handle_data(context, data):
current_50etf_price = data.current('510050.SH', 'close') # 假设 data.current 可以获取最新价格
atm_call_option = context.get_atm_option_contract(
underlying='510050.SH',
strike_year_month=202312,
underlying_price=current_50etf_price,
option_type=OptionCP.CALL
)
if atm_call_option:
print(f"510050.SH 平值认购期权合约代码: {atm_call_option.symbol}, 执行价格: {atm_call_option.strike}") # 假设 OptionContractData 有 symbol 和 strike 属性
else:
print("未找到 510050.SH 的平值认购期权合约")
get_dominant
@abstractmethod
def get_dominant(self, product_code: Union[str, List[str]]) -> Union[str, Dict[str, str]]: ...
描述:
获取主力合约代码。
参数:
product_code(Union[str, List[str]]): 产品代码，例如'rb'(螺纹钢期货) 或产品代码列表。
返回值:
Union[str, Dict[str, str]]:- 如果
product_code为字符串，则返回主力合约代码字符串。 - 如果
product_code为列表，则返回一个字典，键为产品代码，值为对应的主力合约代码。
- 如果
示例 (获取螺纹钢期货主力合约代码):
def handle_data(context, data):
dominant_rebar = context.get_dominant(product_code='rb')
if dominant_rebar:
print(f"螺纹钢期货主力合约代码: {dominant_rebar}")
else:
print("未找到螺纹钢期货主力合约代码")
示例 (获取螺纹钢和豆粕期货的主力合约代码):
def handle_data(context, data):
dominant_contracts = context.get_dominant(product_code=['rb', 'm'])
if dominant_contracts:
print("主力合约代码:")
for product, dominant_symbol in dominant_contracts.items():
print(f" {product}: {dominant_symbol}")
else:
print("未找到主力合约代码")
current_tick
@abstractmethod
def current_tick(self, symbol: str) -> TickData: ...
描述:
查询最新 Tick 行情数据。
参数:
symbol(str): 标的物代码，例如'000001.SZA'。
返回值:
TickData: Tick 行情数据对象。如果未找到 Tick 数据，则返回None。 (需要导入TickData类型)
示例 (获取 000001.SZ 的最新 Tick 行情):
def handle_data(context, data):
latest_tick = context.current_tick(symbol='000001.SZ')
if latest_tick:
print(f"000001.SZ 最新 Tick 价格: {latest_tick.last_price}, 时间: {latest_tick.datetime}") # 假设 TickData 有 last_price 和 datetime 属性
else:
print("未找到 000001.SZ 的最新 Tick 行情")
get_margin_rate
@abstractmethod
def get_margin_rate(self, symbol: str, account_id: str = "") -> MarginRateData: ...
描述:
查询标的物保证金率。
参数:
symbol(str): 标的物代码。account_id(str, 可选): 账户 ID。如果不指定，则使用默认账户。默认为""。
返回值:
MarginRateData: 保证金率数据对象。如果未找到，则返回None。 (需要导入MarginRateData类型)
示例 (获取 rb2010.SHF 的保证金率):
def handle_data(context, data):
margin_rate_data = context.get_margin_rate(symbol='rb2010.SHF')
if margin_rate_data:
print(f"rb2010.SHF 保证金率: 多仓保证金率 {margin_rate_data.long_margin_ratio}, 空仓保证金率 {margin_rate_data.short_margin_ratio}") # 假设 MarginRateData 有 long_margin_ratio 和 short_margin_ratio 属性
else:
print("未找到 rb2010.SHF 的保证金率信息")
get_margin
@abstractmethod
def get_margin(self, symbol: str, account_id: str = "") -> MarginRateData: ...
描述:
查询标的物保证金。功能可能与 get_margin_rate 相同或类似，具体取决于平台实现。
参数:
symbol(str): 标的物代码。account_id(str, 可选): 账户 ID。如果不指定，则使用默认账户。默认为""。
返回值:
MarginRateData: 保证金数据对象。如果未找到，则返回None。 (需要导入MarginRateData类型)
get_commission
@abstractmethod
def get_commission(self, symbol: str, account_id: str = "") -> CommissionRateData: ...
描述:
查询标的物手续费率。
参数:
symbol(str): 标的物代码。account_id(str, 可选): 账户 ID。如果不指定，则使用默认账户。默认为""。
返回值:
CommissionRateData: 手续费率数据对象。如果未找到，则返回None。 (需要导入CommissionRateData类型)
示例 (获取 000001.SZ 的手续费率):
def handle_data(context, data):
commission_data = context.get_commission(symbol='000001.SZ')
if commission_data:
print(f"000001.SZ 手续费率: {commission_data.commission_rate}") # 假设 CommissionRateData 有 commission_rate 属性
else:
print("未找到 000001.SZ 的手续费率信息")
9. 风险控制设置 (Risk Management Settings)
set_commission
@abstractmethod
def set_commission(self, equities_commission: Union[float, None] = None, futures_commission: Union[float, None] = None, options_commission: Union[float, None] = None, account_id: str = ""): ...
描述:
[回测专用] 设置回测手续费模型。
参数:
equities_commission(Union[float, None], 可选): 股票手续费率。如果为None，则使用默认模型。futures_commission(Union[float, None], 可选): 期货手续费率。如果为None，则使用默认模型。options_commission(Union[float, None], 可选): 期权手续费率。如果为None，则使用默认模型。account_id(str, 可选): 账户 ID。如果不指定，则使用默认账户。默认为""。
示例 (回测中自定义股票和期货手续费率):
def initialize(context):
context.set_commission(equities_commission=0.0003, futures_commission=10.0) # 假设 0.0003 为股票 0.03% 手续费率， 10.0 为期货每手 10 元手续费
def handle_data(context, data):
# ... 回测逻辑 ...
pass
set_margin
@abstractmethod
def set_margin(self, symbol: str, margin_ratio: float, account_id: str = ""): ...
描述:
[回测专用] 设置标的物保证金比例。
参数:
symbol(str): 标的物代码。margin_ratio(float): 保证金比例，例如0.1表示 10% 保证金。account_id(str, 可选): 账户 ID。如果不指定，则使用默认账户。默认为""。
示例 (回测中设置 rb2010.SHF 的保证金比例为 15%):
def initialize(context):
context.set_margin(symbol='rb2010.SHF', margin_ratio=0.15)
def handle_data(context, data):
# ... 回测逻辑 ...
pass
set_slippage
@abstractmethod
def set_slippage(self, tick_slippage: Union[Any, None] = None, bar_slippage: Union[Any, None] = None, us_equities: Union[Any, None] = None, account_id: str = ""): ...
描述:
[回测专用] 设置回测滑点模型。
参数:
tick_slippage(Union[Any, None], 可选): Tick 数据滑点模型。模型类型取决于具体实现。如果为None，则使用默认模型。bar_slippage(Union[Any, None], 可选): Bar 数据滑点模型。模型类型取决于具体实现。如果为None，则使用默认模型。us_equities(Union[Any, None], 可选): 美国股票滑点模型 (可能已废弃或为兼容旧版本保留参数)。如果为None，则使用默认模型。account_id(str, 可选): 账户 ID。如果不指定，则使用默认账户。默认为""。
注意: 滑点模型的具体类型和参数设置需要参考平台更详细的文档或示例。
set_slippage_value
@abstractmethod
def set_slippage_value(self, slippage_type: Union[str, None] = None, slippage_value: float = 0, volume_limit: Union[int, None] = None, account_id: str = ""): ...
描述:
[回测专用] 设置固定滑点值。
参数:
slippage_type(Union[str, None], 可选): 滑点类型，例如'fixed_value','percentage'等。具体类型取决于平台实现。如果为None，则使用默认类型。slippage_value(float, 可选): 滑点值。具体含义取决于slippage_type。默认为0。volume_limit(Union[int, None], 可选): 成交量限制。可能用于设置不同成交量范围的滑点值。如果为None，则无成交量限制。account_id(str, 可选): 账户 ID。如果不指定，则使用默认账户。默认为""。
注意: 滑点类型的具体取值和滑点值的含义需要参考平台更详细的文档或示例。
10. 回测设置 (Backtest Settings)
set_vmatch_at
@abstractmethod
def set_vmatch_at(self, val: Union[int, VMatchAt]): ...
描述:
[回测专用] 设置 Tick 回测撮合时间模式。用于控制 Tick 回测中订单撮合发生的时机。
参数:
val(Union[int, VMatchAt]): 撮合时间模式。可以是int或VMatchAt枚举值。 (需要导入VMatchAt枚举类型)VMatchAt.CURRENT_BAR (0): 当前 Tick/Bar 时间撮合。VMatchAt.NEXT_BAR_OPEN (1): 下一个 Bar 开盘价撮合。- 其他可选值请参考平台文档。
示例 (设置 Tick 回测为下一个 Bar 开盘价撮合):
def initialize(context):
context.set_vmatch_at(val=VMatchAt.NEXT_BAR_OPEN)
def handle_data(context, data):
# ... Tick 回测逻辑 ...
pass
set_stock_t1
@abstractmethod
def set_stock_t1(self, on: int): ...
描述:
[回测专用] 设置股票 T+1 交易制度。
参数:
on(int):1开启 T+1 制度，0关闭 T+1 制度。
示例 (回测中开启股票 T+1 制度):
def initialize(context):
context.set_stock_t1(on=1)
def handle_data(context, data):
# ... 回测逻辑 ...
pass
set_dividend_reinvestment
@abstractmethod
def set_dividend_reinvestment(self, on: int): ...
描述:
[回测专用] 设置分红再投资。开启后，回测中收到的股票分红会自动用于再投资购买该股票。
参数:
on(int):1开启分红再投资，0关闭分红再投资。
示例 (回测中开启分红再投资):
def initialize(context):
context.set_dividend_reinvestment(on=1)
def handle_data(context, data):
# ... 回测逻辑 ...
pass
11. 工具方法 (Utils)
logger
@property
def logger(self): ...
描述:
获取策略 Logger 对象。用于记录策略运行日志。
返回值:
Logger: Logger 对象。 (需要导入Logger类型)
示例 (使用 Logger 记录日志):
def handle_data(context, data):
context.logger.info("策略开始运行...")
# ... 策略逻辑 ...
context.logger.warning("可能存在风险...")
context.logger.error("发生错误!")
get_error_msg
@abstractmethod
def get_error_msg(self, error_id: int) -> str: ...
描述:
根据错误 ID 获取错误信息字符串。用于获取下单等操作返回的错误代码的详细信息。
参数:
error_id(int): 错误 ID。
返回值:
str: 错误信息字符串。如果错误 ID 未知，可能返回None或空字符串。
示例 (获取下单错误信息):
def handle_data(context, data):
order_id = context.order(symbol='000001.SZ', volume=1000000) # 假设下单数量过大导致错误
if order_id != 0:
error_msg = context.get_error_msg(order_id)
print(f"下单失败，错误代码: {order_id}, 错误信息: {error_msg}")
注意: 本 API 文档为 IContext 接口类的定义文档，具体实现类 StrategyContext 可能会在此基础上进行扩展或调整。 使用时请参考实际平台的 API 文档和示例代码。 部分类型如 RunMode, TradingCalendar, Portfolio, Position, OrderData, OrderReq, TradeData, ContractData, MarginRateData, CommissionRateData, TickData, Logger, VMatchAt, AcctType, Direction, Offset, OrderType, SubscribeFlag, OptionCP 等需要根据实际 bigtrader 平台的定义进行导入和使用。