策略编写指南¶

本文档旨在帮助策略开发者快速掌握 AKQuant 的策略编写方法。

1. 核心概念 (Glossary)¶

对于量化交易的新手，这里有一些基础术语的解释：

Bar (K线): 包含了某一段时间（如1分钟、1天）内的市场行情，主要包含 5 个数据：
- Open: 开盘价
- High: 最高价
- Low: 最低价
- Close: 收盘价
- Volume: 成交量
Strategy (策略): 你的交易机器人。它的核心工作就是不断地看行情 (on_bar)，然后决定买 (buy) 还是卖 (sell)。
Context (上下文): 机器人的“记事本”和“工具箱”。里面记录了当前有多少钱 (cash)、有多少股票 (positions)，也提供了下单的工具。
Position (持仓): 你当前持有的股票或期货数量。正数表示多头（买入持有），负数表示空头（借券卖出）。
Backtest (回测): 历史模拟。用过去的数据来测试你的策略，看看如果过去这么做，能赚多少钱。

2. 策略生命周期¶

一个策略从开始到结束，会经历以下几个阶段：

__init__: Python 对象初始化，适合定义参数。
on_start: 策略启动时调用，必须在此处使用 self.subscribe() 订阅数据，也可在此注册指标。如果是热启动，需注意不要覆盖已恢复的状态。
on_resume: 仅在热启动时调用（在 on_start 之前）。用于处理从快照恢复后的特殊逻辑。
on_bar: 每一根 K 线闭合时触发 (核心交易逻辑)。
on_tick: 每一个 Tick 到达时触发 (高频/盘口策略)。
on_order: 订单状态变化时触发 (如提交、成交、取消)。
on_trade: 收到成交回报时触发。
on_reject: 订单进入 Rejected 状态时触发。
on_session_start / on_session_end: 会话切换时触发。
on_before_trading / on_after_trading: 交易日级钩子。
on_pre_open: 开盘前最后一个合法决策点，适合“盘前信号，本次 open 成交”。
on_daily_rebalance: 交易日调仓钩子（每天最多一次，适合横截面轮动）。
on_portfolio_update: 账户快照变化时触发。
on_error: 用户回调抛异常时触发，默认触发后继续抛出异常。
on_timer: 定时器触发时调用 (需手动注册)。
on_stop: 策略停止时调用，适合进行资源清理或结果统计。
on_train_signal: 滚动训练触发信号 (仅在 ML 模式下触发)。

2.0 回调速查表¶

回调	何时触发	典型用途	示例入口
`on_start`	策略实例启动后	订阅标的、注册指标、初始化运行态资源	`examples/textbook/ch05_strategy.py`
`on_resume`	热启动恢复时，且早于 `on_start`	恢复连接、打印恢复状态、处理快照续跑逻辑	`examples/21_warm_start_demo.py`、`examples/56_functional_warm_start_demo.py`
`on_bar`	每根 Bar 闭合时	主交易逻辑、指标更新、信号计算	`examples/01_quickstart.py`
`on_tick`	每个 Tick 到达时	高频/盘口响应、逐笔监控	`examples/51_class_tick_callbacks_demo.py`
`on_order`	订单状态变化时	跟踪下单生命周期、联动撤单/重置状态	`examples/08_event_callbacks.py`
`on_trade`	收到成交回报时	成交日志、成交后风控、累计统计	`examples/08_event_callbacks.py`
`on_reject`	订单首次进入 `Rejected` 时	记录拒单原因、告警、降级处理	`examples/50_framework_hooks_demo.py`
`on_session_start`	会话切换开始时	日盘/夜盘切换、session 级状态重置	`examples/50_framework_hooks_demo.py`
`on_session_end`	会话切换结束时	收盘后清理、session 结束打点	`examples/50_framework_hooks_demo.py`
`on_before_trading`	本地交易日首次进入 `Normal` 会话	盘前检查、生成交易日级信号	`examples/50_framework_hooks_demo.py`
`on_pre_open`	每个交易日首个常规行情事件前，由框架定时器抢先触发	盘前信号与“本次 open 成交”语义钩子；可用于表达集合竞价前的最后决策点，但不等同于券商柜台已支持集合竞价专用委托	`examples/52_pre_open_demo.py`
`on_daily_rebalance`	每个交易日最多一次，与 `on_before_trading` 同阶段	横截面选股、统一调仓	`examples/strategies/05_stock_momentum_rotation_timer.py`
`on_after_trading`	离开 `Normal` 会话时，必要时下一事件补发	日终统计、收盘后清理与归档	`examples/50_framework_hooks_demo.py`
`on_portfolio_update`	账户快照变化时增量触发	监控现金/权益变化、推送 UI 或告警	`examples/50_framework_hooks_demo.py`
`on_error`	任一用户回调抛异常时	记录异常源、决定继续/中断策略	`examples/22_strategy_runtime_config_demo.py`
`on_timer`	定时器到点时	定时调仓、盘前任务、节律性检查	`examples/strategies/07_stock_momentum_rotation_on_timer.py`
`on_stop`	策略停止时	汇总统计、资源释放、打印总结	`examples/textbook/ch05_strategy.py`
`on_expiry`	引擎实际执行到期结算/移除后	处理换月、记录结算、清理失效合约	`examples/49_on_expiry_demo.py`
`on_train_signal`	ML 滚动训练窗口触发时	训练模型、切换待激活模型	`examples/10_ml_walk_forward.py`、`examples/55_functional_ml_walk_forward.py`

其中，on_session_start、on_session_end、on_before_trading、on_after_trading、on_portfolio_update、on_reject 这类框架级钩子，推荐直接运行 examples/50_framework_hooks_demo.py 观察触发顺序与日志输出。如果你的目标是“盘前决策，但希望成交价仍是当日 open”，优先看 examples/52_pre_open_demo.py，不要再用通用 on_timer 模拟该语义。如果你只想先看一份“最常用回调的一站式示例”，优先运行 examples/08_event_callbacks.py；它把 on_start/on_bar/on_order/on_trade/on_reject/on_timer/on_portfolio_update/on_stop 放到了同一个脚本里。如果你要写类风格的 Tick 策略，先看 examples/51_class_tick_callbacks_demo.py；如果你更偏好函数式入口，再看 examples/24_functional_tick_simulation_demo.py。

2.1 回调触发契约¶

对于每个 bar/tick/timer 事件，框架按以下顺序分发回调：

on_order / on_trade（若拒单则额外触发 on_reject）
框架钩子（on_session_*、on_before_trading/on_after_trading、on_portfolio_update）
用户事件回调（on_bar / on_tick / on_timer）

说明：

on_reject 对同一订单 id 只触发一次。
回测中已终态拒单会通过上下文快照 recent_rejected_orders 在下一次事件分发时补发，避免因清理活跃订单导致漏触发。
on_before_trading 在本地交易日首次进入常规交易会话时触发一次；默认回测路径下该会话通常表现为 Continuous。
on_pre_open 在每个交易日的首个常规行情事件前，由框架预注册 timer 先触发一次。
on_daily_rebalance 与 on_before_trading 同一阶段触发，每个交易日最多触发一次。
on_before_trading / on_daily_rebalance 始终按“前一交易日/前一时点信息可见”的语义工作；在这些回调里，get_history()、get_account()、get_portfolio_value() 不应看到当日新 bar 或当日更新后的账户视图。
on_after_trading 在离开常规交易会话时触发；若先跨日再收到事件，会在下一事件补发上一交易日的 on_after_trading。
on_pre_open 内若直接调用 buy/sell/order_target_* 且未显式传 fill_policy，框架会自动解析为 price_basis=open, bar_offset=1, temporal=same_cycle。
这里表达的是框架侧“盘前决策，本次 open 成交”的时序语义，不等同于交易所或券商柜台已经实现了集合竞价专用报单、撤单窗口控制或专有价格类型。
新股/新债打新不属于 on_pre_open 或当前统一 submit_order(...) 的默认承诺范围；若要支持，通常需要补齐 broker 专有字段与业务路由。
若需要更精确的交易日边界触发，可在策略中设置 self.enable_precise_day_boundary_hooks = True；该开关只影响 on_before_trading / on_daily_rebalance / on_after_trading 的触发精度，不改变这些日边界回调中的历史数据与账户快照可见窗口。
on_portfolio_update 采用增量触发：初始化时触发一次，后续仅在订单/成交或持仓相关价格变化时触发。
可通过 self.portfolio_update_eps 过滤微小资产波动（默认 0.0，即不过滤）。
停止阶段会在 on_stop 之前补发待触发的 on_session_end / on_after_trading。
on_error 参数为 (error, source, payload)，推荐通过 self.error_mode = "raise" | "continue" 控制行为（默认 raise）。self.re_raise_on_error 仍兼容，作为兜底开关。
推荐使用 self.runtime_config = StrategyRuntimeConfig(...) 统一配置上述行为开关。
旧别名字段与 runtime_config 会自动保持同步。

2.1.1 常规事件分发时序¶

sequenceDiagram
    participant Feed as 行情/Timer 事件
    participant FW as Framework Dispatcher
    participant Strategy as 用户策略

    Feed->>FW: bar / tick / timer 到达
    FW->>Strategy: on_order(...)
    FW->>Strategy: on_trade(...)
    alt 订单首次变为 Rejected
        FW->>Strategy: on_reject(...)
    end
    FW->>Strategy: on_session_start / on_session_end
    FW->>Strategy: on_before_trading / on_after_trading
    FW->>Strategy: on_daily_rebalance
    FW->>Strategy: on_portfolio_update
    alt 当前事件是 Bar
        FW->>Strategy: on_bar(bar)
    else 当前事件是 Tick
        FW->>Strategy: on_tick(tick)
    else 当前事件是 Timer
        FW->>Strategy: on_timer(payload)
    end

2.1.2 停止阶段补发时序¶

sequenceDiagram
    participant Engine as Engine Stop Phase
    participant FW as Framework Dispatcher
    participant Strategy as 用户策略

    Engine->>FW: _on_stop_internal()
    alt 仍有未补发的 session 结束事件
        FW->>Strategy: on_session_end(...)
    end
    alt 仍有未补发的交易日结束事件
        FW->>Strategy: on_after_trading(...)
    end
    FW->>Strategy: on_stop()

2.1.3 何时使用 `on_pre_open`¶

推荐使用 on_pre_open 的场景：

你在集合竞价、盘前扫描、开盘前风控检查后形成信号。
你希望订单默认按“本次 open 成交”建模，而不是等下一根 bar。
你需要一个语义明确的框架钩子，而不是自己维护一套定时器协议。

不推荐继续用通用 on_timer 代替的场景：

需要严格表达“盘前信号，本次 open 成交”。
需要让团队成员一眼看懂策略意图。

三者区别：

on_before_trading: 交易日级语义钩子，强调“这一天开始了”。
on_pre_open: 撮合语义钩子，强调“这是开盘前最后一个可下单点”。
on_timer: 通用调度工具，适合节律任务，不适合承载专门的开盘成交语义。

推荐模板：

class AuctionSignalStrategy(Strategy):
    def __init__(self) -> None:
        self.pending_dates = set()

    def on_start(self) -> None:
        self.subscribe("000001")

    def on_pre_open(self, event: dict[str, object]) -> None:
        trading_date = event["trading_date"]
        if trading_date in self.pending_dates:
            return

        self.pending_dates.add(trading_date)

        signal = self.compute_pre_open_signal()
        if signal > 0:
            # 不显式传 fill_policy 时，默认按当日 open 语义处理
            self.buy("000001", quantity=100)
        elif signal < 0:
            self.sell("000001", quantity=100)

实践建议：

on_pre_open 里尽量只放“最后决策”和“下单”逻辑。
盘前扫描、候选池更新、风控检查可以提前准备，但最终是否成交的决策留在 on_pre_open。
如果你显式传入 fill_policy，将以你的显式配置为准。

时序提醒：

不要把 on_before_trading 当成“同一天稳定先于 on_pre_open 的准备阶段”。
默认路径下，on_pre_open 发生在首个常规事件之前，而 on_before_trading 通常发生在进入 Normal 会话后的首个常规事件上。
即使启用 enable_precise_day_boundary_hooks，on_before_trading 的边界 timer 也不应被当作 on_pre_open 的同日准备链路来依赖。
如果你想做“双阶段”最佳实践，推荐用“前一交易日更晚的回调先准备，下一交易日 on_pre_open 再下单”，例如前一日 on_timer 或 on_after_trading。
参考示例：examples/53_timer_to_pre_open_demo.py。

3. 风险管理 (Risk Management)¶

AKQuant 内置了强大的预交易风控模块，支持在 Engine 层面拦截不合规的订单。

你可以在回测脚本中（Engine 初始化后）配置这些规则：

from akquant import Engine

engine = Engine()
# ... 添加数据 ...

# 获取风控管理器 (注意：PyO3 返回的是副本，修改后需赋值回去)
rm = engine.risk_manager

# 1. 单标的持仓上限 (例如 10%)
# 如果买入导致某标的持仓市值占总权益超过 10%，则拒绝订单
rm.add_max_position_percent_rule(0.10)

# 2. 行业集中度限制 (例如科技股不超过 20%)
# 需要提供 Symbol -> Sector 的映射字典
sector_map = {"AAPL": "Tech", "MSFT": "Tech", "XOM": "Energy"}
rm.add_sector_concentration_rule(0.20, sector_map)

# 3. 总杠杆率熔断 (例如 1.5倍)
# 总敞口 / 总权益 > 1.5 时拒绝开仓
# 对于高杠杆策略，建议关闭默认的现金检查 (check_cash=False)
rm.config.check_cash = False
rm.add_max_leverage_rule(1.5)

# 4. 账户最大回撤限制 (例如 20%)
# 当前权益相对历史峰值回撤超过阈值时，拒绝新订单
rm.add_max_drawdown_rule(0.20)

# 5. 单日亏损限制 (例如 5%)
# 当日权益相对当日首个风控检查时点下跌超过阈值时，拒绝新订单
rm.add_max_daily_loss_rule(0.05)

# 6. 账户净值止损阈值 (例如 80%)
# 当前权益低于“规则首次生效时权益 * 阈值”时，拒绝新订单
rm.add_stop_loss_rule(0.80)

# 应用配置
engine.risk_manager = rm

# 运行回测
engine.run(strategy=MyStrategy)

通过 run_backtest 统一入口也可直接配置账户级风控：

from akquant import run_backtest
from akquant.config import RiskConfig

result = run_backtest(
    data=data,
    strategy=MyStrategy,
    risk_config=RiskConfig(
        max_account_drawdown=0.20,
        max_daily_loss=0.05,
        stop_loss_threshold=0.80,
    ),
)

账户级参数建议（可作为起步值）：

风格	`max_account_drawdown`	`max_daily_loss`	`stop_loss_threshold`
保守	`0.10`	`0.02`	`0.90`
中性	`0.20`	`0.05`	`0.80`
激进	`0.30`	`0.08`	`0.70`

建议先从“中性”起步，再根据策略波动与换手逐步收紧或放宽。

3.1 信用账户回测（融资/融券）¶

若策略需要在回测中启用融资买入或融券卖出，可在 RiskConfig 中显式切换到账户模式 margin：

from akquant.config import RiskConfig

risk_config = RiskConfig(
    account_mode="margin",
    enable_short_sell=True,
    initial_margin_ratio=0.5,
    maintenance_margin_ratio=0.3,
    financing_rate_annual=0.08,
    borrow_rate_annual=0.10,
    allow_force_liquidation=True,
    liquidation_priority="short_first",
)

常用字段说明：

account_mode: "cash" / "margin"。
enable_short_sell: True 时允许股票开空。
initial_margin_ratio: 初始保证金比例（影响可开仓规模）。
maintenance_margin_ratio: 维持担保比例阈值。
allow_force_liquidation: 维持担保比例跌破阈值时是否触发强平。
liquidation_priority: "short_first" 或 "long_first"。

策略中可通过 get_account() 读取信用账户扩展字段：

snap = self.get_account()
print(
    snap["account_mode"],
    snap["borrowed_cash"],
    snap["short_market_value"],
    snap["maintenance_ratio"],
    snap["accrued_interest"],
    snap["daily_interest"],
)

如果策略运行在期货保证金账户语义下，还应重点关注这些字段：

snap["equity"]: 当前账户权益。
snap["used_margin"] / snap["margin"]: 当前已占用保证金。
snap["notional_value"]: 当前期货名义敞口。
snap["unrealized_pnl"]: 当前浮动盈亏。

说明：

期货保证金账户开仓不会像股票现货买入那样扣减全额名义本金。
因此，期货场景下应优先用 equity 观察账户净值变化，用 used_margin 观察保证金占用，用 notional_value 观察杠杆敞口。
如果只需要一个“当前总权益”数值，优先使用 get_portfolio_value()，其口径与 get_account()["equity"] 对齐。

4. 常用工具 (Utilities)¶

AKQuant 提供了一系列便捷工具来简化策略开发。

3.1 日志记录 (Logging)¶

使用 self.log() 可以输出带有当前回测时间戳的日志，方便调试和记录。

def on_bar(self, bar):
    # 自动添加时间戳，例如: [2023-01-01 09:30:00] 信号触发: 买入
    self.log("信号触发: 买入")

    # 支持指定日志级别
    import logging
    self.log("资金不足", level=logging.WARNING)

3.2 便捷数据访问 (Data Access)¶

为了减少代码冗余，Strategy 类提供了当前 Bar/Tick 数据的快捷访问属性：

属性	说明	对应原始代码
`self.symbol`	当前标的代码	`bar.symbol` / `tick.symbol`
`self.close`	当前最新价	`bar.close` / `tick.price`
`self.open`	当前开盘价	`bar.open` (Tick 模式为 0)
`self.high`	当前最高价	`bar.high` (Tick 模式为 0)
`self.low`	当前最低价	`bar.low` (Tick 模式为 0)
`self.volume`	当前成交量	`bar.volume` / `tick.volume`

示例：

def on_bar(self, bar):
    # 旧写法
    if bar.close > bar.open: ...

    # 新写法 (更简洁)
    if self.close > self.open:
        self.buy(self.symbol, 100)

3.3 定时器 (Timer)¶

除了底层的 schedule 方法，AKQuant 提供了更便捷的定时任务注册方式：

add_daily_timer(time_str, payload): 每天在指定时间触发。
- 支持实盘: 在回测模式下预生成所有触发时间；在实盘模式下，每日自动调度下一次触发。
schedule(trigger_time, payload): 在指定时间点（一次性）触发。

def on_start(self):
    # 每天 14:55:00 触发收盘检查
    self.add_daily_timer("14:55:00", "daily_check")

    # 在特定日期时间触发
    self.schedule("2023-01-01 09:30:00", "special_event")

def on_timer(self, payload):
    if payload == "daily_check":
        self.log("Running daily check...")

3.4 横截面策略推荐范式 (Cross-Section Pattern)¶

AKQuant 的 on_bar 按“单事件流”逐条触发。若你要做多标的横截面比较（轮动、排序、打分），推荐优先使用 on_daily_rebalance，由框架保证“每天最多一次”的触发语义。

推荐步骤：

在 on_start 中定义 universe 并订阅标的。
在 on_daily_rebalance 中遍历 universe 计算分数。
在 on_daily_rebalance 中统一选股与调仓。

class CrossSectionStrategy(Strategy):
    def __init__(self, lookback=20):
        self.lookback = lookback
        self.universe = ["sh600519", "sz000858", "sh601318"]
        self.warmup_period = lookback + 1

    def on_start(self):
        for symbol in self.universe:
            self.subscribe(symbol)

    def on_daily_rebalance(self, trading_date, timestamp):
        history_map = self.get_history_map(
            count=self.lookback,
            symbols=self.universe,
            field="close",
        )
        scores = {}
        for symbol, closes in history_map.items():
            if len(closes) < self.lookback:
                continue
            scores[symbol] = (closes[-1] - closes[0]) / closes[0]
        if not scores:
            return
        self.rebalance_to_topn(
            scores=scores,
            top_n=2,
            weight_mode="score",
            long_only=False,
        )

完整示例见：examples/strategies/05_stock_momentum_rotation_timer.py（on_daily_rebalance）与 examples/strategies/07_stock_momentum_rotation_on_timer.py（on_timer 固定时点）。

3.5 横截面方案 B：收齐同 timestamp 后执行¶

当策略没有固定调仓时点（不方便用 on_timer）时，可在 on_bar 中先缓存同一时间片的标的，收齐后再执行一次横截面逻辑。

from collections import defaultdict

class CrossSectionBucketStrategy(Strategy):
    def __init__(self, lookback=20):
        self.lookback = lookback
        self.universe = ["sh600519", "sz000858", "sh601318"]
        self.warmup_period = lookback + 1
        self.pending = defaultdict(set)

    def on_bar(self, bar):
        self.pending[bar.timestamp].add(bar.symbol)
        if len(self.pending[bar.timestamp]) < len(self.universe):
            return
        self.pending.pop(bar.timestamp, None)
        scores = {}
        for symbol in self.universe:
            closes = self.get_history(count=self.lookback, symbol=symbol, field="close")
            if len(closes) < self.lookback:
                return
            scores[symbol] = (closes[-1] - closes[0]) / closes[0]
        best = max(scores, key=lambda s: scores[s])
        self.order_target_percent(target_percent=0.95, symbol=best)

完整示例见：examples/strategies/06_stock_momentum_rotation_bucket.py。

3.6 方案选型对照 (A vs B)¶

维度	方案 A：`on_timer` 统一执行	方案 B：收齐 `timestamp` 后执行
触发方式	固定时点触发（如 14:55）	事件驱动，时间片收齐触发
稳健性	高，不依赖到达顺序	中，需维护缓存并处理缺失
实现复杂度	低，逻辑集中	中，需管理 `timestamp -> symbols`
适用场景	日频/定时调仓、生产默认	无固定调仓时点的横截面策略
常见风险	定时器时间与数据频率不匹配	某些标的缺失导致不触发

建议：优先使用方案 A；只有在无法定义稳定调仓时点时再采用方案 B。

3.7 横截面常见坑位清单¶

停牌/缺失数据：某些标的当日无 Bar 时，方案 B 可能不触发；可设置超时降级，或允许“有效样本数达阈值”即执行。
Universe 漂移：成分股调整后若仍用旧列表，会出现权重与真实池不一致；建议定期刷新并记录生效日期。
调仓时点与成交策略错配：例如 fill_policy={"price_basis":"open","bar_offset":1} 时，收盘时点信号会在下一根撮合；应结合 fill_policy.temporal 明确 timer 是否当期成交。
历史长度不足：新上市或停牌恢复标的数据窗口不完整；评分前统一做 len(closes) 检查并跳过不足样本。
仓位未收敛：多标的先卖后买若资金未及时释放，可能导致买入不足；可采用目标仓位 API 并在下一时点二次收敛。

完整上线检查可参考：横截面策略实战清单。

5. 策略风格选择¶

AKQuant 提供了两种风格的策略开发接口：

风格选择建议可参考：策略风格决策指南。

特性	类风格 (推荐)	函数风格
定义方式	继承 `akquant.Strategy`	定义 `initialize` + `on_bar`（必选），可选 `on_start` / `on_stop` / `on_tick` / `on_order` / `on_trade` / `on_timer`
适用场景	复杂策略、需要维护内部状态、生产环境	快速原型验证、迁移 Zipline/Backtrader 策略
代码结构	面向对象，逻辑封装性好	脚本化，简单直观
API 调用	`self.buy()`, `self.ctx`	`ctx.buy()`, `ctx` 作为参数传递

5.1 函数式回调触发前提¶

回调	触发前提	说明
`on_bar(ctx, bar)`	回测数据流产生 Bar 事件	函数式策略的必选主回调
`on_start(ctx)`	回测启动时触发	对齐类策略 `on_start` 生命周期
`on_stop(ctx)`	回测结束时触发	对齐类策略 `on_stop` 生命周期
`on_tick(ctx, tick)`	回测数据流产生 Tick 事件	仅 Bar 数据集不会触发 Tick 回调
`on_order(ctx, order)`	策略上下文中观察到订单状态变化	每轮事件循环中先于主事件回调触发
`on_trade(ctx, trade)`	`recent_trades` 中出现成交回报	框架会进行成交去重，避免重复触发
`on_expiry(ctx, event)`	引擎实际执行到期结算/移除	仅在 `expiry_date` 驱动的结算真正发生后触发，且触发时账户状态已更新
`on_pre_open(ctx, event)`	每个交易日首个常规行情事件前，由框架预注册 timer 抢先触发	适合函数式“盘前决策，本次 open 成交”场景
`on_timer(ctx, payload)`	已注册的定时器到点触发	支持单次定时与每日定时 payload

5.2 类风格 vs 函数式回调对照¶

类风格	函数式	说明	推荐示例
`on_start(self)`	`on_start(ctx)`	生命周期起点，两种风格都支持	`examples/08_event_callbacks.py`、`examples/23_functional_callbacks_demo.py`
`on_stop(self)`	`on_stop(ctx)`	生命周期终点，两种风格都支持	`examples/textbook/ch05_strategy.py`
`on_bar(self, bar)`	`on_bar(ctx, bar)`	主策略入口，两种风格都支持	`examples/01_quickstart.py`、`examples/23_functional_callbacks_demo.py`
`on_tick(self, tick)`	`on_tick(ctx, tick)`	Tick 事件入口，两种风格都支持	`examples/51_class_tick_callbacks_demo.py`、`examples/24_functional_tick_simulation_demo.py`
`on_order(self, order)`	`on_order(ctx, order)`	订单状态回调，两种风格都支持	`examples/08_event_callbacks.py`
`on_trade(self, trade)`	`on_trade(ctx, trade)`	成交回报回调，两种风格都支持	`examples/08_event_callbacks.py`
`on_expiry(self, event)`	`on_expiry(ctx, event)`	到期结算回调，两种风格都支持	`examples/49_on_expiry_demo.py`
`on_pre_open(self, event)`	`on_pre_open(ctx, event)`	盘前开盘语义回调，两种风格都支持	`examples/52_pre_open_demo.py`
`on_timer(self, payload)`	`on_timer(ctx, payload)`	定时器回调，两种风格都支持	`examples/08_event_callbacks.py`、`examples/23_functional_callbacks_demo.py`
`on_resume(self)`	`on_resume(ctx)`	热启动恢复钩子，两种风格都支持；仅在从快照恢复时触发，且先于 `on_start`	`examples/21_warm_start_demo.py`、`examples/56_functional_warm_start_demo.py`
`on_reject(self, order)`	`on_reject(ctx, order)`	拒单回调，两种风格都支持	`examples/08_event_callbacks.py`、`examples/50_framework_hooks_demo.py`
`on_session_start(self, session, timestamp)`	`on_session_start(ctx, session, timestamp)`	会话边界钩子，两种风格都支持	`examples/50_framework_hooks_demo.py`
`on_session_end(self, session, timestamp)`	`on_session_end(ctx, session, timestamp)`	会话边界钩子，两种风格都支持	`examples/50_framework_hooks_demo.py`
`on_before_trading(self, trading_date, timestamp)`	`on_before_trading(ctx, trading_date, timestamp)`	交易日前边界钩子，两种风格都支持	`examples/50_framework_hooks_demo.py`
`on_after_trading(self, trading_date, timestamp)`	`on_after_trading(ctx, trading_date, timestamp)`	交易日后边界钩子，两种风格都支持	`examples/50_framework_hooks_demo.py`
`on_daily_rebalance(self, trading_date, timestamp)`	`on_daily_rebalance(ctx, trading_date, timestamp)`	交易日调仓钩子，两种风格都支持	`examples/strategies/05_stock_momentum_rotation_timer.py`
`on_portfolio_update(self, snapshot)`	`on_portfolio_update(ctx, snapshot)`	账户快照回调，两种风格都支持	`examples/50_framework_hooks_demo.py`
`on_error(self, error, source, payload)`	`on_error(ctx, error, source, payload)`	用户异常回调，两种风格都支持	`examples/22_strategy_runtime_config_demo.py`
`on_train_signal(self, context)`	`on_train_signal(ctx)`	ML 滚动训练钩子，两种风格都支持；仅在 ML 滚动训练窗口触发时调用	`examples/10_ml_walk_forward.py`、`examples/55_functional_ml_walk_forward.py`

建议：

如果你只是做快速原型，且只依赖 on_bar/on_tick/on_order/on_trade/on_timer，函数式入口通常更轻量。
如果你的目标是“盘前信号，本次 open 成交”，函数式入口现在也可以直接使用 on_pre_open(ctx, event)。
如果你使用 checkpoint 热启动，函数式入口现在也支持 on_resume(ctx)，适合恢复外部连接或非持久化资源，参考 examples/56_functional_warm_start_demo.py。
如果你做 ML walk-forward，函数式入口现在也支持 on_train_signal(ctx)，可用于自定义训练或仅记录训练窗口，参考 examples/55_functional_ml_walk_forward.py。
如果你偏好脚本式策略，函数式入口现在也支持 on_reject/on_session_*/on_before_trading/on_after_trading/on_daily_rebalance/on_portfolio_update 这批框架级钩子。

5.3 相关示例¶

函数式回调基础示例：examples/23_functional_callbacks_demo.py
函数式 Tick 回调模拟示例：examples/24_functional_tick_simulation_demo.py
LiveRunner 支持函数式入口与多 slot 编排：LiveRunner(strategy_cls=on_bar, strategy_id="alpha", strategies_by_slot={"beta": OtherStrategy}, initialize=..., on_tick=..., on_order=..., on_trade=..., on_expiry=..., on_timer=...)
回测多 slot 与策略级风控映射建议使用集中式 BacktestConfig(strategy_config=StrategyConfig(...))：docs/zh/advanced/multi_strategy_guide.md
broker_live 函数式下单示例：examples/39_live_broker_submit_order_demo.py
broker_live 默认执行语义为 strict，可通过 gateway_options={"execution_semantics_mode": "strict"} 显式声明
函数式多策略 slot + 风控示例：examples/40_functional_multi_slot_risk_demo.py
LiveRunner 多策略 slot 编排示例：examples/41_live_multi_slot_orchestration_demo.py
运行后可分别观察输出标记：
- done_functional_callbacks_demo
- done_functional_tick_simulation_demo

6. 编写类风格策略 (Class-based)¶

这是 AKQuant 推荐的策略编写方式，结构清晰，易于扩展。

6.1 数据预热 (Warmup Period)¶

在计算技术指标（如 MA, RSI）时，需要一定长度的历史数据。AKQuant 提供了 warmup_period 机制来自动处理数据预加载。

静态设置 (推荐): 在类中定义 warmup_period = N。
动态设置: 在 __init__ 中设置 self.warmup_period = N。
自动推断: 如果使用内置指标，框架会尝试自动计算所需长度（但显式设置更安全）。

6.2 历史数据获取¶

self.get_history(count, ...): 返回 numpy.ndarray，性能最高，适合计算指标。
self.get_history_df(count, ...): 返回 pandas.DataFrame，包含 OHLCV，适合复杂分析。

6.3 完整示例¶

from akquant import Strategy, Bar
import numpy as np

class MyStrategy(Strategy):
    # 声明需要的预热数据长度 (例如 20日均线需要至少 20 根 Bar)
    warmup_period = 20

    def __init__(self, ma_window=20):
        # 注意: Strategy 类使用了 __new__ 进行初始化，子类不再需要调用 super().__init__()
        self.ma_window = ma_window
        # 如果参数影响预热长度，可以动态覆盖
        self.warmup_period = ma_window + 5

    def on_start(self):
        # 显式订阅数据
        self.subscribe("600000")

    def on_bar(self, bar: Bar):
        # 1. 获取历史数据
        # 返回 numpy array: [close_t-N, ..., close_t-1, close_t]
        history = self.get_history(count=self.ma_window, symbol=bar.symbol, field="close")

        # 检查数据是否足够 (虽然 warmup_period 会保证，但防御性编程是好习惯)
        if len(history) < self.ma_window:
            return

        # 计算均线
        ma_value = np.mean(history)

        # 2. 交易逻辑
        # 获取当前持仓 (使用 Position Helper 或 get_position)
        pos = self.get_position(bar.symbol)

        if bar.close > ma_value and pos == 0:
            self.buy(symbol=bar.symbol, quantity=100)
        elif bar.close < ma_value and pos > 0:
            self.sell(symbol=bar.symbol, quantity=100)

7. 订单与交易详解 (Orders & Execution)¶

7.1 订单生命周期¶

在 AKQuant 中，订单状态流转如下：

New: 订单对象被创建。
Submitted: 订单已发送给交易所/仿真撮合引擎。
Accepted: (实盘模式) 交易所确认接收订单。
Filled: 订单全部成交。
- PartiallyFilled: 订单部分成交（filled_quantity < quantity）。
Cancelled: 订单已取消。
Rejected: 订单被风控或交易所拒绝 (如资金不足、超出涨跌停)。

broker_live + CTP 默认使用严格执行语义：终态以 OnRtnOrder 为准。也就是说，发送撤单请求后，只有收到 OnRtnOrder(Cancelled) 才会进入 Cancelled；收到错误回报后，也需要后续订单回报来最终确认 Rejected。

7.2 常用交易指令¶

市价单 (Market Order):
- self.buy(symbol, quantity)
- self.sell(symbol, quantity)
- 以当前市场最优价格立即成交，保证成交速度，不保证价格。
限价单 (Limit Order):
- self.buy(symbol, quantity, price=10.5)
- 只有当市场价格 <= 10.5 时才买入。
目标仓位 (Target Order):
- self.order_target(target=100, symbol="AAPL"): 调整持仓数量至 100 股。
- self.order_target_percent(target_percent=0.5, symbol="AAPL"): 调整持仓至总资产的 50%。
- self.order_target_value(target_value=10000, symbol="AAPL"): 调整持仓至 10000 元市值。
- self.order_target_weights(target_weights={"AAPL":0.4,"MSFT":0.3}, liquidate_unmentioned=True, rebalance_tolerance=0.01): 按多标的权重统一调仓。
  - 默认权重和不超过 1.0，如需超过请设置 allow_leverage=True。
  - 该接口仍然更偏向 long-only 组合管理；如需正负目标仓位，请优先使用 order_target_positions()。
  - 同周期调仓已升级为 reduce-first 语义：先执行释放约束的腿，再执行增加约束的腿。
- self.order_target_positions(target_positions={"IF2406": -2, "510300": 1000}, liquidate_unmentioned=True): 按多标的目标持仓数量统一调仓，支持正负仓位。
  - allow_short 默认按当前执行环境自动推断；在 cash 或 broker 未声明支持做空时，负目标会被明确拒绝。
  - missing_price_mode="ignore" | "skip" | "fail" 可控制 price_map 缺项时的处理方式。
  - 可通过 self.get_last_target_positions_plan() 查看最近一次调仓计划，确认哪些腿进入了 reduce / increase、哪些腿被跳过或拒绝。

def on_timer(self, payload: str):
    weights = {"sh600519": 0.35, "sz000858": 0.25, "sh601318": 0.20}
    self.order_target_weights(
        target_weights=weights,
        liquidate_unmentioned=True,
        rebalance_tolerance=0.01,
    )

def on_timer(self, payload: str):
    self.order_target_positions(
        target_positions={"IF2406": -2, "510300": 1000},
        liquidate_unmentioned=True,
        allow_short=True,
        missing_price_mode="fail",
    )

    plan = self.get_last_target_positions_plan()
    print(plan["status"], plan["submitted_legs"], plan["skipped_legs"])

7.2.1 显式开平语义与实盘能力¶

AKQuant 当前的订单语义已经不再是单纯的 side，而是 side + position_effect：

buy() / sell() 默认使用 position_effect="auto"，进入执行前会按当前净仓自动拆成 close + open。
short() 默认等价于 side="Sell", position_effect="open"。
cover() 默认等价于 side="Buy", position_effect="close"。
高级场景可直接使用：
- self.submit_order(..., position_effect="open")
- self.submit_order(..., position_effect="close")
- self.submit_order(..., position_effect="close_today")
- self.submit_order(..., position_effect="close_yesterday")

在 broker_live + CTP 路径中，这些语义会继续映射到底层柜台 offset：

open
close
close_today
close_yesterday

你可以通过 self.get_execution_capabilities() 查询当前执行环境是否支持：

position_effect
position_details
supports_short_sell
account_mode

对于 broker_live 环境，返回结果里通常还会包含更细的 broker 能力字段，例如：

broker_name
supported_position_effects
broker_extra_fields

对不支持显式开平或做空的 broker，AKQuant 会在下单前直接给出明确错误，而不是静默降级。

撤单 (Cancel Order):
- self.cancel_order(order_id): 撤销指定订单。
- self.cancel_all_orders(): 撤销当前所有未成交订单。

7.3 OCO 与 Bracket 助手¶

AKQuant 提供了两组交易助手，减少策略中手写订单联动逻辑：

self.create_oco_order_group(first_order_id, second_order_id, group_id=None)
- 把两个订单绑定为 OCO（One-Cancels-the-Other）。
- 任一订单成交后，另一订单会自动撤销。
self.place_bracket_order(symbol, quantity, entry_price=None, stop_trigger_price=None, take_profit_price=None, ...)
- 一次性提交 Bracket 结构。
- 进场单成交后，自动挂出止损/止盈；当止损与止盈同时存在时自动绑定 OCO。

from akquant import OrderStatus, Strategy

class BracketHelperStrategy(Strategy):
    def __init__(self):
        self.entry_order_id = ""

    def on_bar(self, bar):
        if self.get_position(bar.symbol) > 0 or self.entry_order_id:
            return

        self.entry_order_id = self.place_bracket_order(
            symbol=bar.symbol,
            quantity=100,
            stop_trigger_price=bar.close * 0.98,
            take_profit_price=bar.close * 1.04,
            entry_tag="entry",
            stop_tag="stop",
            take_profit_tag="take",
        )

    def on_order(self, order):
        if order.id == self.entry_order_id and order.status in (
            OrderStatus.Cancelled,
            OrderStatus.Rejected,
        ):
            self.entry_order_id = ""

7.4 Trailing Stop 助手¶

如果你需要在策略里直接表达“随价格移动的止损线”，可以使用以下助手：

self.place_trailing_stop(symbol, quantity, trail_offset, side="Sell", trail_reference_price=None, ...)
- 触发后按市价执行（StopTrail -> Market）。
self.place_trailing_stop_limit(symbol, quantity, price, trail_offset, side="Sell", trail_reference_price=None, ...)
- 触发后按限价执行（StopTrailLimit -> Limit）。

from akquant import Strategy

class TrailingHelperStrategy(Strategy):
    def __init__(self):
        self.trailing_order_id = ""

    def on_bar(self, bar):
        if self.get_position(bar.symbol) == 0:
            self.buy(bar.symbol, 100)
            self.trailing_order_id = self.place_trailing_stop(
                symbol=bar.symbol,
                quantity=100,
                trail_offset=1.5,
                side="Sell",
                trail_reference_price=bar.close,
                tag="trail-stop",
            )

完整可运行脚本见：examples/36_trailing_orders.py。

6.3 市场规则与 T+1 (Market Rules)¶

在 A 股市场回测中，T+1 交易规则是一个非常重要的限制：当天买入的股票，第二个交易日才能卖出。

启用 T+1¶

默认情况下，AKQuant 使用 T+0 规则（便于美股或期货回测）。如需启用 T+1，请在 run_backtest 中设置：

# 启用 T+1 规则 (适用于 A 股)
akquant.run_backtest(
    ...,
    t_plus_one=True,
    commission_rate=0.0003,
    stamp_tax_rate=0.001  # 配合印花税设置
)

对策略逻辑的影响¶

启用 T+1 后，你需要区分总持仓和可用持仓：

self.get_position(symbol): 返回总持仓（包含今日买入未解锁的部分）。

self.ctx.get_available_position(symbol): 返回可用持仓（即今日可卖出的数量）。 > 推荐使用 Position 辅助类： >

pos = self.position  # 获取当前 symbol 的 Position 对象
print(pos.size)      # 总持仓
print(pos.available) # 可用持仓

示例代码：

def on_bar(self, bar: Bar):
    # 使用 Position Helper
    pos = self.position

    # 卖出逻辑：必须检查可用持仓
    if signal_sell and pos.available > 0:
        self.sell(bar.symbol, pos.available)

注意：如果你在 T+1 模式下尝试卖出超过 available 的数量，订单会被风控模块（Risk Manager）拒绝 (Rejected)，并提示 "Insufficient available position"。

6.4 账户与持仓查询¶

除了 get_position，你还可以查询更多账户信息：

self.ctx.cash: 当前账户可用资金。
self.get_portfolio_value(): 当前账户总权益。
self.get_account(): 当前账户快照；在保证金账户中可进一步读取 used_margin、notional_value、unrealized_pnl、maintenance_ratio 等字段。
self.get_trades(): 获取历史所有已平仓交易记录（Closed Trades）。
self.get_open_orders(): 获取当前未成交订单。

on_trade 与 get_trades() 的语义不同：

on_trade(self, trade) 接收的是当前事件步内的增量成交回报（适合做实时响应）。
self.get_trades() 返回的是累计“已平仓”交易（Closed Trades），未平仓时不会出现在这个列表里。

推荐模式：

class MyStrategy(Strategy):
    def __init__(self):
        self.recent_exec_count = 0

    def on_trade(self, trade):
        self.recent_exec_count += 1
        print("incremental trade:", trade.order_id, trade.symbol, trade.quantity)

    def on_stop(self):
        closed = self.get_trades()
        print("closed trades:", len(closed))

6.5 标的静态属性查询（推荐）¶

当你需要在策略中读取行权价、到期日、合约乘数、期权类型、标的代码等静态属性时，优先使用策略 API，而不是依赖 bar.extra。

可用接口：

self.get_instrument(symbol): 返回 InstrumentSnapshot。
self.get_instrument_field(symbol, field): 返回单字段值。
self.get_instrument_config(symbol, fields=None): 兼容接口；支持单字段或多字段批量读取。
self.get_instruments(symbols=None): 返回多个标的的快照字典。

这些接口在 on_start 即可使用（回测启动阶段已注入快照）。

import akquant
from akquant import Bar, Strategy


class MetaAwareStrategy(Strategy):
    def on_start(self):
        self.subscribe("OPTION_A")
        expiry = self.get_instrument_field("OPTION_A", "expiry_date")
        strike = self.get_instrument_field("OPTION_A", "strike_price")
        print("meta:", expiry, strike)

    def on_bar(self, bar: Bar):
        meta = self.get_instrument_config(
            bar.symbol, fields=["asset_type", "option_type", "multiplier"]
        )
        if meta["asset_type"] == "OPTION" and meta["option_type"] == "CALL":
            pass

7. 进阶功能¶

7.1 事件回调¶

除了 on_bar，你还可以重写其他回调函数来处理更精细的逻辑：

on_order(self, order): 订单状态更新时触发。
on_trade(self, trade): 订单成交时触发。
on_expiry(self, event): 到期结算事件触发。仅当引擎实际执行到期结算/移除后触发，回调参数为事件字典，常见字段包括 symbol、expiry_date、quantity_closed、cash_flow 与 settlement_type。最小可运行示例见：examples/49_on_expiry_demo.py。

7.2 指标 (Indicators)¶

AKQuant 采用“平台双主流、策略单主流”模式。每个策略需要显式设置 indicator_mode，并使用对应注册接口：

indicator_mode="precompute" + register_precomputed_indicator(...)
indicator_mode="incremental" + register_incremental_indicator(...)

from akquant import Bar, SMA, Strategy

class IndicatorStrategy(Strategy):
    def __init__(self):
        self.indicator_mode = "precompute"
        self.sma20 = SMA(20)
        self.register_precomputed_indicator("sma20", self.sma20)

    def on_start(self):
        self.subscribe("AAPL")

    def on_bar(self, bar: Bar):
        val = self.sma20.get_value(bar.symbol, bar.timestamp)
        if bar.close > val:
            self.buy(bar.symbol, 100)

from akquant import Bar, SMA, Strategy

class IncrementalIndicatorStrategy(Strategy):
    def __init__(self):
        self.indicator_mode = "incremental"
        self.sma20 = SMA(20)
        self.register_incremental_indicator(
            "sma20",
            self.sma20,
            source="close",
            symbols=["AAPL"],
        )

    def on_bar(self, bar: Bar):
        if bar.symbol != "AAPL":
            return
        val = self.sma20.value
        if val is None:
            return
        if bar.close > val:
            self.buy(bar.symbol, 100)

增量模式新增了两项推荐能力：

indicator_factory: 为每个 symbol 创建独立指标实例，适合多标的策略，避免状态串线。
warmup_bars: 在进入正式事件流前，先用 start_time 之前的历史 Bar 预热增量指标。

from akquant import Bar, SMA, Strategy

class MultiSymbolIncrementalStrategy(Strategy):
    def __init__(self):
        self.indicator_mode = "incremental"

    def on_start(self):
        self.register_incremental_indicator(
            "sma20",
            indicator_factory=lambda: SMA(20),
            source="close",
            symbols=["AAPL", "MSFT"],
            warmup_bars=20,
        )

    def on_bar(self, bar: Bar):
        val = self.sma20.value
        if val is None:
            return
        if bar.close > val:
            self.buy(bar.symbol, 100)


说明：

*   单标的旧写法 `register_incremental_indicator("sma20", self.sma20, ...)` 仍然兼容。
*   如果一个共享实例被多个 `symbol` 复用，框架会显式报错，提示改为 `indicator_factory`。
*   `warmup_bars` 只会消费正式开始时间之前的历史数据，不会重复消费第一根有效 Bar。
*   如果你需要编写自己的私有指标，而不仅仅是使用内置 `SMA/EMA`，请继续阅读：[自定义指标指南](./custom_indicator.md)。
## 8. 高级特性：热启动 (Warm Start)


AKQuant 支持**热启动 (Warm Start)** 功能，允许你保存回测状态并在未来恢复。这对于长周期分段回测、滚动训练或模拟实盘环境非常有用。

### 8.1 核心机制

*   **保存快照**: 使用 `save_snapshot` 将引擎状态保存到文件。
*   **恢复运行**: 使用 `run_warm_start` 从快照恢复并继续运行。

### 8.2 策略适配

为了支持热启动，策略类提供了 `on_resume` 生命周期钩子和 `is_restored` 属性。

*   **`on_resume()`**: 仅在从快照恢复时调用（在 `on_start` 之前）。
*   **`self.is_restored`**: 布尔值，指示当前策略实例是否是从快照恢复的。

**示例代码**：

```python
def on_start(self):
    # 1. 初始化指标 (仅在冷启动时)
    if not self.is_restored:
        self.sma = SMA(30)
    else:
        self.log("Resumed from snapshot. Indicators retained.")

    # 2. 注册指标 (必须执行)
    self.register_indicator("sma", self.sma)

    # 3. 订阅行情 (必须执行)
    self.subscribe(self.symbol)

更多详细信息，请参阅热启动指南。 ```