本文汇总执行能力、日志与定时器、仓位管理器以及框架常量与枚举等辅助功能。
执行能力 get_execution_capabilities()
无参数。返回字典,表示当前通道能做什么。回测现券样本如下。
| 参数 | 参数名称 | 参数类型 | 描述 |
|---|---|---|---|
broker_live |
是否实盘通道 | bool |
回测为 False |
client_order_id |
是否接受自定义订单号 | bool |
回测为 False,不要自行填写客户订单号 |
order_type |
是否支持订单类型 | bool |
回测为 True |
time_in_force_str |
是否接受字符串有效期 | bool |
回测为 False |
position_effect |
是否支持开平方向 | bool |
回测为 True |
reduce_only |
是否支持只减仓 | bool |
回测为 True |
position_details |
是否提供持仓明细 | bool |
回测为 False |
account_mode |
账户模式 | str |
现券为 "cash" |
supports_short_sell |
是否支持融券卖空 | bool |
现券为 False |
broker_extra_fields |
通道扩展字段 | list |
扩展字段名列表。回测为 [] |
时间、日志与定时器
日志 log(msg, level=logging.INFO)
输出带策略时间的日志,设置日志级别前需要先导入 logging。
import logging
class MyStrategy(Strategy):
def on_bar(self, bar) -> None:
self.log("信号已触发")
self.log("历史数据不足", logging.WARNING)
self.log("拒单", logging.ERROR)
| 级别 | 语义 | 量化场景示例 |
|---|---|---|
DEBUG |
细粒度诊断 | 排障分支 |
INFO |
正常运行的关键节点 | 订单提交/成交 |
WARNING |
可恢复的降级 | 资金不足拒单 |
ERROR |
操作失败 | 回调异常 |
时间转换
| 方法/属性 | 说明 |
|---|---|
to_local_time(timestamp) |
UTC 纳秒时间戳转为策略本地时间 |
format_time(timestamp, fmt="%Y-%m-%d %H:%M:%S") |
格式化本地时间 |
now |
当前策略本地时间;尚无有效运行时间时可能为 None |
框架时间戳统一使用 UTC 纳秒整数,不要直接把它当作秒级 Unix 时间戳。
定时器
请根据实际情况需要,使用下列接口:
| 方法 | 说明 | 回测 | 实盘 |
|---|---|---|---|
schedule(trigger_time, payload) |
指定时间点一次性触发 | 是 | 是 |
schedule_daily(time_str, payload) |
每个交易日在指定时间触发 | 是 | 是(每日自动调度下一次) |
schedule_weekly(time_str, payload) |
每周首个交易日触发,节假日顺延 | 是 | 交易日历未知时不可用 |
schedule_monthly(time_str, payload) |
每月首个交易日触发,节假日顺延 | 是 | 交易日历未知时不可用 |
from qmfquant import Strategy
class MyStrategy(Strategy):
def on_start(self) -> None:
# 每天 14:55 收盘前检查(回测与实盘都可用)
self.schedule_daily("14:55:00", "daily_check")
# 实盘盘前替代 on_pre_open
self.schedule_daily("09:25:00", "pre_open")
def on_timer(self, payload: str) -> None:
if payload == "pre_open":
self.log("盘前检查")
elif payload == "daily_check":
self.log(f"收盘前检查,当前权益:{self.equity:.2f}")
定时器使用规则:
- 建议在
on_start()中统一注册。 - 多个定时器使用不同
payload,共用一个on_timer()。 - 定时回调中若要下单,应明确传入
symbol,不要依赖self.symbol。 - 定时回调中若要使用价格,应明确确认最新价来源;定时器本身不会创建新的 Bar 或 Tick。
- 横截面/定期调仓优先使用
on_cross_section;schedule_*+on_timer面向通用自定义时点任务。 - 若目的是实盘盘前决策,使用
schedule_daily("09:25:00", ...)或on_before_trading,不要写on_pre_open。
仓位管理器
set_sizer(sizer) 设置的是默认下单数量的算法,不是把持仓目标定死。写在 on_start() 或 __init__() 里各调用一次即可。之后 buy()、short() 不写 quantity 时,按这一次下单时的资金和价格重新计算数量。
| 参数 | 类型 | 说明 |
|---|---|---|
sizer |
Sizer |
必填,下面三种之一 |
返回值:None。
| 类型 | 参数 | 每次省略 quantity 时 |
|---|---|---|
FixedSize(size=100.0) |
size:固定数量 |
始终用这个数量,例如每次 100 股 |
PercentSizer(percents=10.0) |
percents:0 到 100 的百分数,10.0 表示 10%,不是 0.1 |
按当时资金的这个百分比换算数量 |
AllInSizer() |
无 | 按当时可用资金尽量买入 |
写出 quantity 时以写出的数量为准,管理器不参与。sell() 省略数量时卖的是当前可卖多头,cover() 省略数量时平掉当前空头,这两处不用仓位管理器。
from qmfquant import FixedSize, Strategy
class FixedBuyStrategy(Strategy):
def on_start(self) -> None:
self.set_sizer(FixedSize(100))
def on_bar(self, bar) -> None:
if self.get_position(bar.symbol) == 0:
self.buy(bar.symbol)
股票策略仍建议在 buy() 和 sell() 里写出数量,回测和实盘对得上。需要按资金比例调到某个目标仓位时,用 order_target_percent(),不建议用仓位管理器代替。
框架常量与枚举
推荐从 qmfquant 导入后比较,不要比较中文显示文本。
| 枚举 | 常用值 |
|---|---|
OrderSide |
Buy、Sell |
OrderStatus |
New、Submitted、PartiallyFilled、Filled、Cancelled、Rejected、Expired |
OrderType |
Market、Limit、StopMarket、StopLimit |
TimeInForce |
GTC、IOC、FOK、Day |
PositionEffect |
Auto、Open、Close |
AssetType |
Stock、Fund |
TradingSession |
PreOpen、Continuous、CallAuction、Break、Closed |
from qmfquant import Order, OrderSide, OrderStatus, Strategy, TradingSession
class EnumUsageStrategy(Strategy):
def on_order(self, order: Order) -> None:
if order.side == OrderSide.Buy:
self.log("这是一笔买单")
if order.status in {
OrderStatus.Filled,
OrderStatus.Cancelled,
OrderStatus.Rejected,
OrderStatus.Expired,
}:
self.log("订单已经结束")
def on_bar(self, bar) -> None:
if self.ctx.session == TradingSession.Continuous:
pass
TimeInForce.Day:日终结算会把当日仍未成交的 Day 单置为 Expired。若希望“收盘决策、次日成交”的委托不受当日过期约束,可使用默认 GTC。
相关阅读

评论
登录后参与讨论,与站内账号体系共用。