Akquant

AKQuant 量化策略开发指南

任务目标

本 Skill 用于辅助 AI 编程智能体生成符合 akquant 框架规范的可执行量化策略代码。能力包括策略设计、回测配置、订单管理、风控规则、参数优化与横截面策略实现。

核心能力清单

• 策略类生成：继承 Strategy 基类，实现生命周期钩子
• 数据接口配置：准备 DataFrame 数据、设置预热期、访问历史数据
• 事件驱动机制：on_bar/on_tick/on_order/on_trade 等回调
• 订单管理：市价单/限价单/目标仓位/OCO/Bracket/Trailing Stop
• 风控规则：持仓限制/回撤熔断/止损阈值/行业集中度
• 参数优化：网格搜索与滚动优化（Walk-Forward）
• 多策略编排：slot 映射与策略级风控

触发条件

当用户表达以下意图时触发：

• 开发量化交易策略
• 配置回测环境与参数
• 设置风险控制规则
• 进行参数优化与调优
• 实现横截面或轮动策略
• 排查策略运行错误

策略开发工作流

阶段一：理解需求

1. 识别策略类型：趋势跟踪、均值回归、横截面轮动、套利等 2. 确定数据需求：时间周期、标的范围、字段要求 3. 明确风控约束：持仓上限、止损止盈、回撤限制

阶段二：设计策略结构

参考 strategy-patterns.md 选择范式：

• 类风格（推荐）：继承 Strategy，封装状态与逻辑
• 函数风格：initialize + on_bar，快速原型

关键决策点：

• 预热期设置：根据指标窗口长度计算
• 历史数据访问：get_history (numpy) 或 get_history_df (DataFrame)
• 执行模式：NextOpen（下一 Bar 开盘）或 CurrentClose（当前 Bar 收盘）

阶段三：编写策略代码

使用 assets/strategy-template.py 作为起点：

from akquant import Strategy, Bar

class MyStrategy(Strategy):
    warmup_period = 20  # 预热数据长度

    def __init__(self, param1=10):
        self.param1 = param1

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

    def on_bar(self, bar: Bar):
        # 核心交易逻辑
        history = self.get_history(self.param1, bar.symbol, "close")
        if len(history) < self.param1:
            return

        import numpy as np
        ma = np.mean(history)
        pos = self.get_position(bar.symbol)

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

阶段四：配置回测环境

参考 api-reference.md 设置参数：

from akquant import run_backtest

result = run_backtest(
    strategy=MyStrategy,
    data=df,
    symbol="600000",
    initial_cash=500_000.0,
    commission_rate=0.0003,
    stamp_tax_rate=0.001,
    t_plus_one=True,  # A 股 T+1 规则
    warmup_period=20,
    execution_mode="NextOpen",
)

阶段五：设置风控规则

参考 risk-management.md 配置：

from akquant.config import RiskConfig

result = run_backtest(
    ...,
    risk_config=RiskConfig(
        max_position_pct=0.10,  # 单标的持仓不超过 10%
        max_account_drawdown=0.20,  # 最大回撤 20%
        max_daily_loss=0.05,  # 单日亏损 5%
    ),
)

阶段六：参数优化

参考 optimization.md 执行：

from akquant import run_grid_search, run_walk_forward

# 网格搜索
results = run_grid_search(
    strategy=MyStrategy,
    param_grid={"param1": [10, 20, 30]},
    data=df,
    sort_by="sharpe_ratio",
)

# 滚动优化（推荐）
wfo_results = run_walk_forward(
    strategy=MyStrategy,
    param_grid={"param1": [10, 20, 30]},
    data=df,
    train_period=250,
    test_period=60,
    metric="sharpe_ratio",
)

横截面策略开发

参考 cross-section-guide.md 实现多标的轮动：

推荐范式：使用 on_timer 统一触发调仓

class CrossSectionStrategy(Strategy):
    def __init__(self):
        self.universe = ["sh600519", "sz000858", "sh601318"]

    def on_start(self):
        self.add_daily_timer("14:55:00", "rebalance")

    def on_timer(self, payload):
        if payload != "rebalance":
            return

        # 计算所有标分数
        scores = {}
        for symbol in self.universe:
            history = self.get_history(20, symbol, "close")
            scores[symbol] = (history[-1] - history[0]) / history[0]

        # 选出最佳标的并调仓
        best = max(scores, key=scores.get)
        self.order_target_percent(0.95, symbol=best)

资源索引

环境准备与依赖管理

使用 uv 管理项目环境（推荐）

由于 akquant 依赖 pandas>=3.0.0，全局安装可能与现有项目存在版本冲突。推荐使用 uv 创建隔离环境：

1. 安装 uv：若已安装则跳过

# macOS
brew install uv

# Linux
curl -LsSf https://astral.sh/uv/install.sh | sh

# Windows (PowerShell)
powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"

2. 创建项目并初始化环境

# 创建项目目录
mkdir my-quant-strategy
cd my-quant-strategy

# 初始化项目（创建 pyproject.toml）
uv init

# 创建虚拟环境并安装依赖
uv venv
uv add akquant pandas numpy

3. 运行策略脚本

# 方式一：使用 uv run（推荐）
uv run python my_strategy.py

# 方式二：激活虚拟环境后运行
source .venv/bin/activate  # macOS/Linux
# .venv\Scripts\activate   # Windows
python my_strategy.py

4. 依赖版本锁定

uv 会自动生成 uv.lock 文件，确保团队依赖一致：

# 安装精确版本（从 lock 文件）
uv sync

# 添加新依赖
uv add scipy  # 自动更新 lock 文件

5. 项目结构建议

my-quant-strategy/
├── .venv/              # 虚拟环境（uv 自动创建）
├── pyproject.toml      # 项目配置
├── uv.lock             # 依赖锁定文件
├── strategies/         # 策略脚本
│   ├── ma_strategy.py
│   └── cross_section.py
└── data/               # 数据文件
    └── stock_data.csv

快速启动命令

# 一键创建并运行策略项目
mkdir quant-project && cd quant-project
uv init
uv venv
uv add akquant pandas numpy

# 创建策略文件（使用模板）
cat > strategy.py << 'EOF'
from akquant import Strategy, Bar, run_backtest
import pandas as pd
import numpy as np

class MyStrategy(Strategy):
    warmup_period = 20

    def on_bar(self, bar: Bar):
        closes = self.get_history(20, bar.symbol, "close")
        if len(closes) < 20:
            return
        ma = np.mean(closes)
        pos = self.get_position(bar.symbol)
        if bar.close > ma and pos == 0:
            self.buy(bar.symbol, 100)
        elif bar.close < ma and pos > 0:
            self.sell(bar.symbol, pos)

# 准备数据并运行回测
# result = run_backtest(strategy=MyStrategy, data=df, symbol="600000")
EOF

# 运行策略
uv run python strategy.py

注意事项

1. 预热期计算：确保 warmup_period >= 指标所需的最大窗口长度 2. T+1 规则：A 股策略需设置 t_plus_one=True，并区分总持仓与可用持仓 3. 风控优先级：显式参数 > 配置对象 > 默认值 4. 数据格式：DataFrame 必须包含 date/open/high/low/close/volume/symbol 字段 5. 横截面触发：优先使用 on_timer，无固定时点再考虑 timestamp 收齐方案 6. 优化风险：网格搜索易过拟合，推荐使用滚动优化验证稳健性

使用示例

示例 1：双均线策略

from akquant import Strategy, Bar
import numpy as np

class DualMAStrategy(Strategy):
    warmup_period = 30

    def __init__(self, fast=10, slow=20):
        self.fast = fast
        self.slow = slow
        self.warmup_period = slow + 1

    def on_bar(self, bar: Bar):
        fast_ma = np.mean(self.get_history(self.fast, bar.symbol, "close"))
        slow_ma = np.mean(self.get_history(self.slow, bar.symbol, "close"))

        pos = self.get_position(bar.symbol)
        if fast_ma > slow_ma and pos == 0:
            self.buy(bar.symbol, 100)
        elif fast_ma < slow_ma and pos > 0:
            self.sell(bar.symbol, pos)

示例 2：带风控的趋势策略

from akquant import Strategy, Bar, run_backtest
from akquant.config import RiskConfig
import numpy as np

class TrendStrategy(Strategy):
    warmup_period = 20

    def __init__(self, ma_window=20, stop_loss=0.05):
        self.ma_window = ma_window
        self.stop_loss = stop_loss

    def on_bar(self, bar: Bar):
        ma = np.mean(self.get_history(self.ma_window, bar.symbol, "close"))
        pos = self.get_position(bar.symbol)

        if bar.close > ma * 1.02 and pos == 0:
            self.buy(bar.symbol, 100)
        elif bar.close < ma * 0.98 and pos > 0:
            self.sell(bar.symbol, pos)

# 运行回测
result = run_backtest(
    strategy=TrendStrategy,
    data=df,
    symbol="600000",
    initial_cash=1_000_000.0,
    risk_config=RiskConfig(
        max_position_pct=0.20,
        max_account_drawdown=0.15,
        stop_loss_threshold=0.85,
    ),
)

示例 3：横截面动量轮动

from akquant import Strategy, run_backtest
import numpy as np

class MomentumRotation(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)
        self.add_daily_timer("14:55:00", "rebalance")

    def on_timer(self, payload):
        if payload != "rebalance":
            return

        scores = {}
        for symbol in self.universe:
            closes = self.get_history(self.lookback, symbol, "close")
            if len(closes) < self.lookback:
                return
            scores[symbol] = (closes[-1] - closes[0]) / closes[0]

        # 选出最佳标的，持仓 95%
        best = max(scores, key=scores.get)
        self.order_target_percent(0.95, symbol=best)

API 速查手册

本文档提供 akquant 核心 API 的快速参考，涵盖回测入口、策略基类、交易对象与配置类。

目录

• 回测入口
• 策略基类
• 交易对象
• 配置类
• 数据对象

---

回测入口

run_backtest

最常用的回测入口函数，封装了引擎的初始化和配置过程。

from akquant import run_backtest

result = run_backtest(
    strategy=MyStrategy,          # 策略类或实例
    data=df,                      # DataFrame 或 Dict[str, DataFrame]
    symbol="600000",              # 标的代码或列表
    initial_cash=500_000.0,       # 初始资金
    commission_rate=0.0003,       # 佣金率
    stamp_tax_rate=0.001,         # 印花税率
    transfer_fee_rate=0.00002,    # 过户费率
    min_commission=5.0,           # 最低佣金
    slippage=0.0001,              # 滑点
    volume_limit_pct=0.25,        # 成交量限制
    execution_mode="NextOpen",    # 执行模式
    t_plus_one=False,             # T+1 规则
    warmup_period=20,             # 预热期
    start_time="2023-01-01",      # 开始时间
    end_time="2023-12-31",        # 结束时间
    risk_config=RiskConfig(...),  # 风控配置
)

关键参数说明：

执行模式说明：

• NextOpen：信号触发后在下一 Bar 开盘价成交（默认）
• CurrentClose：信号触发后在当前 Bar 收盘价成交

run_grid_search

网格搜索参数优化。

from akquant import run_grid_search

results = run_grid_search(
    strategy=MyStrategy,
    param_grid={"ma_window": [10, 20, 30], "stop_loss": [0.03, 0.05]},
    data=df,
    sort_by="sharpe_ratio",
    ascending=False,
    timeout=60.0,  # 单次任务超时（秒）
    result_filter=lambda m: m["trade_count"] >= 50,  # 结果筛选
)

run_walk_forward

滚动优化（Walk-Forward Optimization）。

from akquant import run_walk_forward

wfo_results = run_walk_forward(
    strategy=MyStrategy,
    param_grid={"ma_window": [10, 20, 30]},
    data=df,
    train_period=250,   # 训练窗口长度
    test_period=60,     # 测试窗口长度
    metric="sharpe_ratio",
)

---

策略基类

Strategy

策略基类，用户应继承此类并重写回调方法。

生命周期钩子

属性访问

交易方法

# 市价单
self.buy(symbol, quantity)
self.sell(symbol, quantity)

# 限价单
self.buy(symbol, quantity, price=10.5)
self.sell(symbol, quantity, price=10.5)

# 止损/止盈单
self.buy(symbol, quantity, trigger_price=9.5)  # 价格跌破 9.5 时买入

# 目标仓位
self.order_target(target=100, symbol="AAPL")  # 调整持仓至 100 股
self.order_target_percent(0.5, symbol="AAPL")  # 调整持仓至总资产 50%
self.order_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,    # 容忍小偏差
)

# 撤单
self.cancel_order(order_id)
self.cancel_all_orders(symbol)

# OCO 订单组
self.create_oco_order_group(first_order_id, second_order_id)

# Bracket 订单
self.place_bracket_order(
    symbol=bar.symbol,
    quantity=100,
    stop_trigger_price=bar.close * 0.98,
    take_profit_price=bar.close * 1.04,
)

# 跟踪止损
self.place_trailing_stop(
    symbol=bar.symbol,
    quantity=100,
    trail_offset=1.5,  # 跟踪偏移
    side="Sell",
)

数据查询方法

# 获取历史数据（numpy array）
history = self.get_history(count=20, symbol="AAPL", field="close")

# 获取历史数据（DataFrame）
df = self.get_history_df(count=20, symbol="AAPL")

# 获取持仓
position = self.get_position(symbol)  # 总持仓
available = self.get_available_position(symbol)  # 可用持仓（T+1）

# 获取所有持仓
positions = self.get_positions()  # Dict[str, float]

# 获取账户信息
cash = self.get_cash()
account = self.get_account()  # {"cash": ..., "equity": ..., "market_value": ...}

# 获取订单与成交
order = self.get_order(order_id)
open_orders = self.get_open_orders(symbol)
trades = self.get_trades()

工具方法

# 日志记录
self.log("信号触发", level=logging.INFO)

# 定时任务
self.add_daily_timer("14:55:00", "rebalance")  # 每日定时
self.schedule("2023-01-01 09:30:00", "special")  # 单次定时

# 订阅数据
self.subscribe("AAPL")

# 时间转换
local_time = self.to_local_time(timestamp)
time_str = self.format_time(timestamp, "%Y-%m-%d %H:%M:%S")

---

交易对象

Order

@dataclass
class Order:
    id: str                        # 订单 ID
    symbol: str                    # 标的代码
    side: OrderSide                # Buy / Sell
    order_type: OrderType          # Market / Limit / StopMarket
    status: OrderStatus            # New / Filled / Cancelled / Rejected
    quantity: float                # 委托数量
    filled_quantity: float         # 已成交数量
    price: Optional[float]         # 委托价格
    average_filled_price: float    # 成交均价
    trigger_price: Optional[float] # 触发价格
    time_in_force: TimeInForce     # GTC / IOC / FOK / Day
    created_at: int                # 创建时间戳（纳秒）
    updated_at: int                # 更新时间戳
    tag: str                       # 标签
    reject_reason: str             # 拒绝原因

Trade

单次成交记录（一个订单可能对应多次成交）。

@dataclass
class Trade:
    id: str           # 成交 ID
    order_id: str     # 对应订单 ID
    symbol: str       # 标的代码
    side: OrderSide   # 方向
    quantity: float   # 成交数量
    price: float      # 成交价格
    commission: float # 手续费
    timestamp: int    # 成交时间戳

ClosedTrade

已平仓交易记录（开仓+平仓完整周期）。

@dataclass
class ClosedTrade:
    entry_time: int       # 开仓时间
    exit_time: int        # 平仓时间
    entry_price: float    # 开仓价格
    exit_price: float     # 平仓价格
    quantity: float       # 数量
    pnl: float            # 盈亏金额
    return_pct: float     # 收益率
    duration: int         # 持仓时间（秒）
    mae: float            # 最大不利变动
    mfe: float            # 最大有利变动

---

配置类

BacktestConfig

from akquant import BacktestConfig, StrategyConfig

config = BacktestConfig(
    strategy_config=StrategyConfig(
        initial_cash=500_000.0,
        commission_rate=0.0003,
        t_plus_one=True,
    ),
    start_time="2023-01-01",
    end_time="2023-12-31",
    instruments=["600000"],
    show_progress=True,
)

StrategyConfig

from akquant import StrategyConfig

strategy_config = StrategyConfig(
    initial_cash=500_000.0,
    commission_rate=0.0003,
    stamp_tax_rate=0.001,
    slippage=0.0001,
    volume_limit_pct=0.25,
    t_plus_one=True,
    max_long_positions=10,
    risk=RiskConfig(
        max_position_pct=0.10,
        max_account_drawdown=0.20,
    ),
)

RiskConfig

from akquant.config import RiskConfig

risk_config = RiskConfig(
    active=True,
    safety_margin=0.0001,
    max_order_size=10000,           # 单笔最大委托数量
    max_order_value=100_000.0,      # 单笔最大委托金额
    max_position_size=50000,        # 单标的最大持仓数量
    max_position_pct=0.10,          # 单标的持仓上限（占总权益）
    sector_concentration=0.20,      # 行业集中度限制
    max_account_drawdown=0.20,      # 账户最大回撤
    max_daily_loss=0.05,            # 单日最大亏损
    stop_loss_threshold=0.80,       # 账户净值止损阈值
)

InstrumentConfig

from akquant import InstrumentConfig

# 股票配置
stock_config = InstrumentConfig(
    symbol="600000",
    asset_type="STOCK",
    lot_size=100,
)

# 期货配置
futures_config = InstrumentConfig(
    symbol="IF2312",
    asset_type="FUTURES",
    multiplier=300.0,      # 合约乘数
    margin_ratio=0.12,     # 保证金率
    tick_size=0.2,         # 最小变动价位
)

# 期权配置
option_config = InstrumentConfig(
    symbol="10003720",
    asset_type="OPTION",
    option_type="CALL",
    strike_price=3.0,
    expiry_date="2023-12-27",
    underlying_symbol="510050",
)

---

数据对象

Bar

K 线数据对象。

@dataclass
class Bar:
    timestamp: int               # Unix 时间戳（纳秒）
    open: float                  # 开盘价
    high: float                  # 最高价
    low: float                   # 最低价
    close: float                 # 收盘价
    volume: float                # 成交量
    symbol: str                  # 标的代码
    extra: Dict[str, float]      # 扩展数据字典
    
    @property
    def timestamp_str(self) -> str:
        """时间字符串"""
        ...

Tick

Tick 数据对象。

@dataclass
class Tick:
    timestamp: int     # Unix 时间戳（纳秒）
    price: float       # 最新价
    volume: float      # 成交量
    symbol: str        # 标的代码

---

BacktestResult

回测结果对象。

@dataclass
class BacktestResult:
    metrics_df: pd.DataFrame       # 绩效指标表
    trades_df: pd.DataFrame        # 平仓交易表
    orders_df: pd.DataFrame        # 委托记录表
    executions_df: pd.DataFrame    # 成交流水表
    positions_df: pd.DataFrame     # 每日持仓表
    equity_curve: List[Tuple]      # 权益曲线
    
    # 方法
    def exposure_df(self, freq="D") -> pd.DataFrame:
        """组合暴露分解"""
        ...
    
    def attribution_df(self, by="symbol") -> pd.DataFrame:
        """归因分析"""
        ...
    
    def capacity_df(self, freq="D") -> pd.DataFrame:
        """容量代理指标"""
        ...

常用指标：

横截面策略指南

本文档详细说明 akquant 框架中横截面策略的开发方法，包括触发机制、数据准备、调仓逻辑与常见陷阱。

目录

• 横截面策略概述
• 推荐范式：on_timer
• 备选方案：timestamp 收齐
• 实战清单
• 常见陷阱

---

横截面策略概述

什么是横截面策略？

横截面策略是在同一时间点对多个标的进行比较、排序或打分，然后根据结果进行调仓。常见应用场景：

• 行业轮动：选择强势行业
• 因子选股：根据因子得分选股
• 动量轮动：选择近期表现最好的标的

核心特点

• 同时点比较：必须在同一时间点对多个标的进行评估
• 相对排名：关注的是标的之间的相对表现，而非绝对收益
• 定期调仓：通常在固定时点触发调仓

关键挑战

1. 触发时机：如何确保所有标的数据就绪后再执行 2. 数据同步：处理停牌、缺失数据等情况 3. 调仓效率：先卖后买，避免资金占用问题

---

推荐范式：on_timer

设计思路

使用 on_timer 在固定时点统一触发调仓，确保：

• 所有标的在同一时刻被评估
• 调仓逻辑集中在一个地方，易于管理
• 避免 on_bar 逐条触发导致的状态不一致

基本结构

from akquant import Strategy
import numpy as np

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)
        
        # 注册每日定时器
        self.add_daily_timer("14:55:00", "rebalance")
    
    def on_timer(self, payload):
        if payload != "rebalance":
            return
        
        # 1. 计算所有标分数
        scores = {}
        for symbol in self.universe:
            closes = self.get_history(self.lookback, symbol, "close")
            if len(closes) < self.lookback:
                return  # 数据不足，跳过本次调仓
            scores[symbol] = (closes[-1] - closes[0]) / closes[0]
        
        # 2. 选出最佳标的
        best = max(scores, key=scores.get)
        
        # 3. 调仓
        self.order_target_percent(0.95, symbol=best)

完整示例：动量轮动

from akquant import Strategy, run_backtest
import numpy as np

class MomentumRotation(Strategy):
    """
    动量轮动策略：
    - 每日收盘前计算所有标的的动量分数
    - 持有分数最高的标的
    """
    
    def __init__(self, lookback=20, top_k=1):
        self.lookback = lookback
        self.top_k = top_k
        self.universe = ["sh600519", "sz000858", "sh601318", "sh601166"]
        self.warmup_period = lookback + 1
    
    def on_start(self):
        # 订阅所有标的
        for symbol in self.universe:
            self.subscribe(symbol)
        
        # 每日 14:55 触发调仓
        self.add_daily_timer("14:55:00", "rebalance")
    
    def on_timer(self, payload):
        if payload != "rebalance":
            return
        
        # 计算动量分数
        scores = {}
        for symbol in self.universe:
            closes = self.get_history(self.lookback, symbol, "close")
            
            # 数据不足则跳过
            if len(closes) < self.lookback:
                self.log(f"{symbol} 数据不足，跳过本次调仓")
                return
            
            # 动量 = (最新价 - N日前价格) / N日前价格
            momentum = (closes[-1] - closes[0]) / closes[0]
            scores[symbol] = momentum
        
        # 选出 Top K 标的
        sorted_symbols = sorted(scores, key=scores.get, reverse=True)
        selected = sorted_symbols[:self.top_k]
        
        # 计算目标权重（等权）
        weight_per_stock = 0.95 / self.top_k
        target_weights = {s: weight_per_stock for s in selected}
        
        # 调仓
        self.order_target_weights(
            target_weights=target_weights,
            liquidate_unmentioned=True,  # 平掉未选中的持仓
            rebalance_tolerance=0.01,    # 容忍 1% 偏差
        )
        
        self.log(f"调仓完成: {selected}")

# 运行回测
result = run_backtest(
    strategy=MomentumRotation,
    data=data_dict,  # Dict[str, DataFrame]
    symbol=list(data_dict.keys()),
    initial_cash=1_000_000.0,
    commission_rate=0.0003,
    t_plus_one=True,
)

多因子打分

class MultiFactorStrategy(Strategy):
    def __init__(self):
        self.universe = ["sh600519", "sz000858", "sh601318"]
        self.warmup_period = 30
    
    def on_start(self):
        for symbol in self.universe:
            self.subscribe(symbol)
        self.add_daily_timer("14:55:00", "rebalance")
    
    def on_timer(self, payload):
        if payload != "rebalance":
            return
        
        scores = {}
        for symbol in self.universe:
            # 因子 1：动量
            closes = self.get_history(20, symbol, "close")
            if len(closes) < 20:
                return
            momentum = (closes[-1] - closes[0]) / closes[0]
            
            # 因子 2：波动率（反向）
            returns = np.diff(closes) / closes[:-1]
            volatility = np.std(returns)
            
            # 因子 3：成交量相对变化
            volumes = self.get_history(20, symbol, "volume")
            if len(volumes) < 20:
                return
            vol_ratio = volumes[-1] / np.mean(volumes)
            
            # 综合得分
            score = (
                momentum * 0.5 -           # 动量权重 50%
                volatility * 10 +          # 波动率权重（负向）
                vol_ratio * 0.3            # 成交量权重 30%
            )
            scores[symbol] = score
        
        # 选出最佳标的
        best = max(scores, key=scores.get)
        self.order_target_percent(0.95, symbol=best)

---

备选方案：timestamp 收齐

适用场景

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

实现方式

from collections import defaultdict

class BucketStrategy(Strategy):
    def __init__(self, lookback=20):
        self.lookback = lookback
        self.universe = ["sh600519", "sz000858", "sh601318"]
        self.warmup_period = lookback + 1
        self.pending = defaultdict(set)  # timestamp -> set of symbols
    
    def on_bar(self, bar):
        # 1. 记录当前时间片已到达的标的
        self.pending[bar.timestamp].add(bar.symbol)
        
        # 2. 检查是否所有标的都已到达
        if len(self.pending[bar.timestamp]) < len(self.universe):
            return  # 未收齐，等待
        
        # 3. 收齐后执行横截面逻辑
        self.pending.pop(bar.timestamp, None)
        
        scores = {}
        for symbol in self.universe:
            closes = self.get_history(self.lookback, symbol, "close")
            if len(closes) < self.lookback:
                return
            scores[symbol] = (closes[-1] - closes[0]) / closes[0]
        
        best = max(scores, key=scores.get)
        self.order_target_percent(0.95, symbol=best)

方案对比

推荐：优先使用 on_timer；只有在无法定义稳定调仓时点时再采用 timestamp 收齐方案。

---

实战清单

设计阶段

• [ ] 明确横截面触发机制：优先 on_timer，无固定时点再考虑时间片收齐方案
• [ ] 明确信号时点与成交时点关系：特别是 execution_mode="next_open" 的跨 Bar 成交
• [ ] 固定并版本化 universe 来源，记录成分生效日期与调仓周期
• [ ] 定义持仓约束：单标的上限、行业集中度、现金留存、最小交易单位

数据阶段

• [ ] 统一时区、交易日历和缺失值策略，避免横截面混入异步样本
• [ ] 评分前校验窗口长度，跳过历史不足样本并统计样本覆盖率
• [ ] 对停牌/复牌、涨跌停、成交量异常设置可追踪的降级处理
• [ ] 固化数据快照和拉取参数，保证回测可复现

执行阶段

• [ ] 采用目标仓位接口进行调仓，减少先卖后买导致的仓位漂移
• [ ] 设置调仓容忍区间，避免小幅分数波动引发高换手
• [ ] 对订单拒绝进行集中监控，重点关注 orders_df.reject_reason
• [ ] 定期二次收敛仓位，处理一次调仓未完全达到目标的问题

风控阶段

• [ ] 启用账户级风控：max_account_drawdown、max_daily_loss、stop_loss_threshold
• [ ] 结合策略特性设置限额：max_position_pct、max_order_value 等
• [ ] 建立风控触发后的策略行为约定：降仓、暂停、仅平仓
• [ ] 保留风控触发日志，便于复盘和参数迭代

验证阶段

• [ ] 做时序切分验证（滚动窗口/分阶段）而不是只看全样本结果
• [ ] 观察关键稳定性指标：换手率、持仓集中度、滑点敏感性、容量约束
• [ ] 对比不同执行模式与调仓频率，确认收益来源稳定
• [ ] 将参数与结果快照写入实验记录，便于回溯

上线前检查

• [ ] 清单通过：触发、数据、执行、风控、验证五项全部打勾
• [ ] 演练异常场景：缺行情、拒单、延迟触发、交易日切换
• [ ] 固定运行配置与依赖版本，避免环境漂移
• [ ] 准备回滚方案：参数回滚、策略停用、版本回退

---

常见陷阱

1. 停牌/缺失数据导致不触发

# ❌ 问题：某标的停牌，on_timer 仍能触发，但该标的数据缺失
def on_timer(self, payload):
    for symbol in self.universe:
        closes = self.get_history(20, symbol, "close")
        # 如果某标的停牌，closes 长度可能不足
        if len(closes) < 20:
            return  # 直接返回，导致所有标的都无法调仓

# ✅ 解决：跳过缺失标的，继续处理其他标的
def on_timer(self, payload):
    scores = {}
    for symbol in self.universe:
        closes = self.get_history(20, symbol, "close")
        if len(closes) < 20:
            self.log(f"{symbol} 数据不足，跳过")
            continue  # 跳过，继续处理其他标的
        scores[symbol] = (closes[-1] - closes[0]) / closes[0]
    
    if len(scores) < len(self.universe) * 0.8:  # 有效样本 < 80%
        return  # 样本不足，跳过本次调仓
    
    # 正常调仓
    best = max(scores, key=scores.get)
    self.order_target_percent(0.95, symbol=best)

2. Universe 漂移

# ❌ 问题：使用硬编码的 universe，未考虑成分调整
self.universe = ["sh600519", "sz000858", "sh601318"]

# ✅ 解决：定期刷新 universe 并记录生效日期
def on_start(self):
    self.load_universe()  # 从文件或数据库加载
    self.add_daily_timer("09:30:00", "refresh_universe")

def on_timer(self, payload):
    if payload == "refresh_universe":
        self.load_universe()

3. 调仓时点与执行模式错配

# ❌ 问题：14:55 触发调仓，但使用 NextOpen 模式
# 实际成交在次日开盘，信号与成交时点不一致
self.add_daily_timer("14:55:00", "rebalance")

# ✅ 解决：明确记录信号时点与成交时点
def on_timer(self, payload):
    self.log(f"信号时点: {self.now}")
    # NextOpen 模式下，实际成交在次日开盘
    self.order_target_percent(...)

4. 历史长度不足

# ❌ 问题：新上市或停牌恢复标的数据窗口不完整
def on_timer(self, payload):
    for symbol in self.universe:
        closes = self.get_history(20, symbol, "close")
        score = (closes[-1] - closes[0]) / closes[0]  # 可能只有 10 根数据

# ✅ 解决：统一做长度检查
def on_timer(self, payload):
    for symbol in self.universe:
        closes = self.get_history(20, symbol, "close")
        if len(closes) < 20:
            continue  # 跳过不足样本
        score = (closes[-1] - closes[0]) / closes[0]

5. 仓位未收敛

# ❌ 问题：多标的先卖后买，资金未及时释放，导致买入不足
def on_timer(self, payload):
    for symbol in self.universe:
        self.order_target_percent(0.25, symbol)  # 每个标的 25%

# ✅ 解决：使用 order_target_weights 统一调仓
def on_timer(self, payload):
    weights = {s: 0.25 for s in self.universe}
    self.order_target_weights(
        target_weights=weights,
        liquidate_unmentioned=True,
        rebalance_tolerance=0.01,  # 容忍小偏差
    )

6. timestamp 收齐方案中的缺失问题

# ❌ 问题：某标的永远缺失，pending 永远无法收齐
self.pending[bar.timestamp].add(bar.symbol)
if len(self.pending[bar.timestamp]) < len(self.universe):
    return  # 永远等待

# ✅ 解决：设置超时或降级阈值
def on_bar(self, bar):
    self.pending[bar.timestamp].add(bar.symbol)
    
    # 超时检查：如果下一个 timestamp 已到达，强制处理上一个
    if len(self.pending) > 1:
        old_ts = min(self.pending.keys())
        symbols = self.pending.pop(old_ts)
        if len(symbols) >= len(self.universe) * 0.8:  # 80% 到达即可执行
            self.execute_cross_section(old_ts)

---

最佳实践

1. 触发机制：优先使用 on_timer，确保同一时点统一决策 2. 数据检查：每次评分前检查历史数据长度，跳过不足样本 3. 样本阈值：设置有效样本率阈值（如 80%），避免数据缺失导致决策偏差 4. 调仓接口：使用 order_target_weights 统一调仓，避免先卖后买问题 5. 风控集成：横截面策略必须配置风控，防止单一标的过度集中 6. 日志记录：记录每次调仓的信号、决策与成交，便于复盘 7. 异常演练：测试停牌、涨跌停、拒单等异常场景的表现

参数优化指南

本文档详细说明 akquant 框架的参数优化方法，包括网格搜索、滚动优化与参数模型驱动。

目录

• 优化方法对比
• 网格搜索
• 滚动优化
• 参数模型驱动
• 最佳实践

---

优化方法对比

推荐：生产策略必须使用滚动优化验证稳健性。

---

网格搜索

基本用法

from akquant import run_grid_search

# 定义参数网格
param_grid = {
    "ma_window": [10, 20, 30],
    "stop_loss": [0.03, 0.05, 0.08],
}

# 运行网格搜索
results = run_grid_search(
    strategy=MyStrategy,
    param_grid=param_grid,
    data=df,
    sort_by="sharpe_ratio",
    ascending=False,
)

# 查看结果
print(results.head())

多目标排序

results = run_grid_search(
    strategy=MyStrategy,
    param_grid=param_grid,
    data=df,
    sort_by=["sharpe_ratio", "total_return"],  # 优先夏普，其次收益
    ascending=[False, False],
)

结果筛选

def result_filter(metrics):
    return (
        metrics.get("trade_count", 0) >= 50 and        # 交易次数 >= 50
        metrics.get("sharpe_ratio", 0) > 1.0 and       # 夏普 > 1.0
        metrics.get("max_drawdown_pct", 1.0) < 0.2     # 最大回撤 < 20%
    )

results = run_grid_search(
    strategy=MyStrategy,
    param_grid=param_grid,
    data=df,
    result_filter=result_filter,
)

动态预热期

def warmup_calc(params):
    # 预热期 = 指标窗口 + 缓冲
    return params["ma_window"] + 5

results = run_grid_search(
    strategy=MyStrategy,
    param_grid=param_grid,
    data=df,
    warmup_calc=warmup_calc,
)

参数约束

def param_constraint(params):
    # 短期均线 < 长期均线
    return params["fast_window"] < params["slow_window"]

results = run_grid_search(
    strategy=MyStrategy,
    param_grid={
        "fast_window": [5, 10, 15],
        "slow_window": [20, 30, 60],
    },
    constraint=param_constraint,
)

资源控制

results = run_grid_search(
    strategy=MyStrategy,
    param_grid=param_grid,
    data=df,
    timeout=60.0,            # 单次任务超时 60 秒
    max_tasks_per_child=1,   # 每次任务重启进程
)

断点续传

results = run_grid_search(
    strategy=MyStrategy,
    param_grid=param_grid,
    data=df,
    db_path="optimization.db",  # 持久化到 SQLite
)

特性：

• 实时保存：每完成一个组合立即写入数据库
• 断点续传：程序中断后自动跳过已完成的组合
• 结果复用：可从数据库读取历史结果

---

滚动优化

原理

滚动优化（Walk-Forward Optimization, WFO）模拟真实的时间流逝，将数据切分为多个 [训练集 | 测试集] 窗口：

窗口 1: [训练 1-3 月] → [测试 4 月]
窗口 2: [训练 2-4 月] → [测试 5 月]
窗口 3: [训练 3-5 月] → [测试 6 月]
...
最终: 拼接所有测试段结果

核心思想：永远只用过去的数据来决定现在的参数。

基本用法

from akquant import run_walk_forward

wfo_results = run_walk_forward(
    strategy=MyStrategy,
    param_grid=param_grid,
    data=df,
    train_period=250,      # 训练窗口长度（Bar 数量）
    test_period=60,        # 测试窗口长度（Bar 数量）
    metric="sharpe_ratio", # 优化目标
    ascending=False,
)

print(wfo_results)

多目标优化

wfo_results = run_walk_forward(
    strategy=MyStrategy,
    param_grid=param_grid,
    data=df,
    train_period=250,
    test_period=60,
    metric=["sharpe_ratio", "total_return"],  # 多目标
    ascending=[False, False],
)

完整示例

from akquant import Strategy, run_walk_forward
import numpy as np

class MAStrategy(Strategy):
    def __init__(self, fast=10, slow=20):
        self.fast = fast
        self.slow = slow
        self.warmup_period = slow + 1
    
    def on_bar(self, bar):
        fast_ma = np.mean(self.get_history(self.fast, bar.symbol, "close"))
        slow_ma = np.mean(self.get_history(self.slow, bar.symbol, "close"))
        
        pos = self.get_position(bar.symbol)
        if fast_ma > slow_ma and pos == 0:
            self.buy(bar.symbol, 100)
        elif fast_ma < slow_ma and pos > 0:
            self.sell(bar.symbol, pos)

# 定义参数网格
param_grid = {
    "fast": [5, 10, 15],
    "slow": [20, 30, 60],
}

# 参数约束
def constraint(params):
    return params["fast"] < params["slow"]

# 动态预热期
def warmup_calc(params):
    return params["slow"] + 1

# 运行滚动优化
wfo_results = run_walk_forward(
    strategy=MAStrategy,
    param_grid=param_grid,
    data=df,
    train_period=250,
    test_period=60,
    metric="sharpe_ratio",
    constraint=constraint,
    warmup_calc=warmup_calc,
    initial_cash=500_000.0,
)

参数选择

---

参数模型驱动

定义参数模型

from akquant import ParamModel, IntParam, FloatParam

class StrategyParams(ParamModel):
    fast_period: int = IntParam(10, ge=2, le=200, title="快线周期")
    slow_period: int = IntParam(30, ge=3, le=500, title="慢线周期")
    stop_loss: float = FloatParam(0.05, ge=0.01, le=0.20, title="止损比例")

策略类声明

from akquant import Strategy

class MyStrategy(Strategy):
    PARAM_MODEL = StrategyParams
    
    def __init__(self, fast_period=10, slow_period=30, stop_loss=0.05):
        self.fast_period = fast_period
        self.slow_period = slow_period
        self.stop_loss = stop_loss
        self.warmup_period = slow_period + 1

导出 Schema

from akquant import get_strategy_param_schema

schema = get_strategy_param_schema(MyStrategy)
# 返回 JSON Schema，用于前端表单生成

校验参数

from akquant import validate_strategy_params

# 校验用户输入
params = validate_strategy_params(
    MyStrategy,
    {"fast_period": 12, "slow_period": 36, "stop_loss": 0.08},
)

# 校验失败会抛出异常
try:
    validate_strategy_params(
        MyStrategy,
        {"fast_period": 300, "slow_period": 20},  # fast > max, fast > slow
    )
except Exception as e:
    print(f"参数校验失败: {e}")

与网格搜索结合

# PARAM_MODEL 用于单次回测校验
# param_grid 用于网格搜索

results = run_grid_search(
    strategy=MyStrategy,
    param_grid={
        "fast_period": [5, 10, 15],
        "slow_period": [20, 30, 60],
    },
    data=df,
)

---

最佳实践

1. 避免过拟合

# ❌ 错误：直接使用网格搜索最优参数
results = run_grid_search(...)
best_params = results.iloc[0].to_dict()
# 在生产中使用 best_params → 可能是过拟合的产物

# ✅ 正确：使用滚动优化验证稳健性
wfo_results = run_walk_forward(...)
# 如果 WFO 结果稳定，说明参数具有稳健性

2. 设置合理的筛选条件

def result_filter(metrics):
    return (
        metrics.get("trade_count", 0) >= 30 and        # 交易次数足够
        metrics.get("sharpe_ratio", 0) > 0.5 and       # 夏普为正
        metrics.get("max_drawdown_pct", 1.0) < 0.3 and # 回撤可控
        metrics.get("win_rate", 0) > 0.4               # 胜率合理
    )

3. 使用参数约束减少无效组合

def constraint(params):
    # 逻辑约束
    if params["fast"] >= params["slow"]:
        return False
    
    # 合理性约束
    if params["stop_loss"] > params["take_profit"]:
        return False
    
    return True

4. 动态预热期避免数据不足

def warmup_calc(params):
    # 预热期 = 最大指标窗口 + 缓冲
    max_window = max(params["fast"], params["slow"], params["rsi_period"])
    return max_window + 10

5. 分析参数稳定性

# 网格搜索后分析参数敏感性
results = run_grid_search(...)

# 查看同一参数不同值的表现
for ma in [10, 20, 30]:
    subset = results[results["ma_window"] == ma]
    print(f"MA={ma}: 平均夏普={subset['sharpe_ratio'].mean():.2f}")

# 如果某参数值的表现在显著优于其他，可能过拟合

6. 分阶段验证

# 阶段 1：网格搜索探索参数空间
grid_results = run_grid_search(
    strategy=MyStrategy,
    param_grid={"ma": range(5, 50, 5)},  # 粗粒度
    data=df_train,
)

# 阶段 2：在候选范围内细粒度搜索
top_ma = grid_results.iloc[0]["ma"]
fine_results = run_grid_search(
    strategy=MyStrategy,
    param_grid={"ma": range(top_ma - 5, top_ma + 6, 1)},  # 细粒度
    data=df_validate,
)

# 阶段 3：滚动优化最终验证
wfo_results = run_walk_forward(
    strategy=MyStrategy,
    param_grid={"ma": [top_ma - 2, top_ma, top_ma + 2]},
    data=df_test,
)

---

完整示例

from akquant import (
    Strategy,
    run_grid_search,
    run_walk_forward,
    ParamModel,
    IntParam,
    FloatParam,
    validate_strategy_params,
)
import numpy as np

# 1. 定义参数模型
class TrendParams(ParamModel):
    fast: int = IntParam(10, ge=2, le=100, title="快线")
    slow: int = IntParam(30, ge=5, le=200, title="慢线")
    stop_loss: float = FloatParam(0.05, ge=0.01, le=0.20, title="止损")

# 2. 定义策略
class TrendStrategy(Strategy):
    PARAM_MODEL = TrendParams
    
    def __init__(self, fast=10, slow=30, stop_loss=0.05):
        self.fast = fast
        self.slow = slow
        self.stop_loss = stop_loss
        self.warmup_period = slow + 1
        self.entry_price = None
    
    def on_bar(self, bar):
        # 指标计算
        fast_ma = np.mean(self.get_history(self.fast, bar.symbol, "close"))
        slow_ma = np.mean(self.get_history(self.slow, bar.symbol, "close"))
        
        pos = self.get_position(bar.symbol)
        
        # 开仓
        if fast_ma > slow_ma and pos == 0:
            self.buy(bar.symbol, 100)
            self.entry_price = bar.close
        
        # 止损
        elif pos > 0 and self.entry_price:
            pnl = (bar.close - self.entry_price) / self.entry_price
            if pnl < -self.stop_loss:
                self.sell(bar.symbol, pos)
                self.entry_price = None
        
        # 平仓
        elif fast_ma < slow_ma and pos > 0:
            self.sell(bar.symbol, pos)
            self.entry_price = None

# 3. 参数约束
def constraint(params):
    return params["fast"] < params["slow"]

# 4. 动态预热期
def warmup_calc(params):
    return params["slow"] + 5

# 5. 结果筛选
def result_filter(metrics):
    return (
        metrics.get("trade_count", 0) >= 20 and
        metrics.get("sharpe_ratio", 0) > 0.5
    )

# 6. 网格搜索（探索阶段）
param_grid = {
    "fast": [5, 10, 15, 20],
    "slow": [30, 40, 50, 60],
    "stop_loss": [0.03, 0.05, 0.08],
}

grid_results = run_grid_search(
    strategy=TrendStrategy,
    param_grid=param_grid,
    data=df,
    constraint=constraint,
    warmup_calc=warmup_calc,
    result_filter=result_filter,
    sort_by="sharpe_ratio",
    db_path="optimization.db",
)

print("网格搜索结果 Top 5:")
print(grid_results.head())

# 7. 滚动优化（验证阶段）
wfo_results = run_walk_forward(
    strategy=TrendStrategy,
    param_grid={
        "fast": [10, 15],
        "slow": [30, 40],
        "stop_loss": [0.05],
    },
    data=df,
    train_period=250,
    test_period=60,
    metric="sharpe_ratio",
    constraint=constraint,
    warmup_calc=warmup_calc,
)

print("\n滚动优化结果:")
print(wfo_results)

风控配置指南

本文档详细说明 akquant 框架的风险管理机制，包括预交易风控、账户级风控与策略级风控。

目录

• 风控概述
• RiskConfig 配置
• Engine 层风控
• 账户级风控
• 策略级风控
• 最佳实践

---

风控概述

akquant 内置了强大的预交易风控模块，支持在 Engine 层面拦截不合规的订单。风控系统采用"检查链"模式，每个规则独立判断，任一规则拒绝则订单被拦截。

风控层级

订单提交 → 单笔检查 → 持仓检查 → 账户检查 → 策略检查 → 提交成功
           ↓           ↓           ↓           ↓
         拒绝         拒绝        拒绝        拒绝

---

RiskConfig 配置

基本配置

from akquant.config import RiskConfig

risk_config = RiskConfig(
    active=True,                    # 是否启用风控
    safety_margin=0.0001,           # 安全垫
    max_order_size=10000,           # 单笔最大委托数量
    max_order_value=100_000.0,      # 单笔最大委托金额
    max_position_size=50000,        # 单标的最大持仓数量
    max_position_pct=0.10,          # 单标的持仓上限（占总权益）
    sector_concentration=0.20,      # 行业集中度限制
    restricted_list=["ST001", "ST002"],  # 禁止交易标的
    max_account_drawdown=0.20,      # 账户最大回撤
    max_daily_loss=0.05,            # 单日最大亏损
    stop_loss_threshold=0.80,       # 账户净值止损阈值
)

通过 run_backtest 配置

from akquant import run_backtest
from akquant.config import RiskConfig

result = run_backtest(
    strategy=MyStrategy,
    data=df,
    risk_config=RiskConfig(
        max_position_pct=0.10,
        max_account_drawdown=0.20,
    ),
)

通过 BacktestConfig 配置

from akquant import BacktestConfig, StrategyConfig, run_backtest
from akquant.config import RiskConfig

config = BacktestConfig(
    strategy_config=StrategyConfig(
        initial_cash=500_000.0,
        risk=RiskConfig(
            max_position_pct=0.10,
            max_account_drawdown=0.20,
        ),
    ),
)

result = run_backtest(strategy=MyStrategy, data=df, config=config)

---

Engine 层风控

获取 RiskManager

from akquant import Engine

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

rm = engine.risk_manager

单标的持仓上限

# 某标的持仓市值不超过总权益的 10%
rm.add_max_position_percent_rule(0.10)

行业集中度限制

# 单行业持仓不超过总权益的 20%
sector_map = {
    "AAPL": "Tech",
    "MSFT": "Tech",
    "XOM": "Energy",
    "JPM": "Finance",
}
rm.add_sector_concentration_rule(0.20, sector_map)

杠杆率熔断

# 总敞口 / 总权益 > 1.5 时拒绝开仓
rm.config.check_cash = False  # 关闭现金检查（高杠杆策略）
rm.add_max_leverage_rule(1.5)

应用配置

# 修改后必须赋值回去
engine.risk_manager = rm

---

账户级风控

账户级风控在订单提交前检查账户状态，触发后拒绝新的下单请求。

最大回撤限制

risk_config = RiskConfig(
    max_account_drawdown=0.20,  # 最大回撤 20%
)

规则说明：

• 以历史权益峰值为基准
• 当前权益回撤超过阈值后，新的下单请求被拒绝
• 已有持仓不受影响，仅阻止新开仓

单日亏损限制

risk_config = RiskConfig(
    max_daily_loss=0.05,  # 单日亏损 5%
)

规则说明：

• 以当日首次风控检查时的权益为基准
• 当日亏损超过阈值后，新的下单请求被拒绝
• 次日重置基准

账户净值止损

risk_config = RiskConfig(
    stop_loss_threshold=0.80,  # 跌至初始权益 80% 时触发
)

规则说明：

• 当前权益 < 规则首次生效时权益 × 阈值时触发
• 触发后新的下单请求被拒绝
• 永久生效，不会重置

参数建议

---

策略级风控

多策略场景下，支持为每个策略设置独立的风控规则。

配置策略级风控

from akquant import BacktestConfig, StrategyConfig, run_backtest

config = BacktestConfig(
    strategy_config=StrategyConfig(
        strategy_id="alpha",
        strategies_by_slot={"beta": BetaStrategy},
        
        # 策略级持仓限制
        strategy_max_order_size={"alpha": 10, "beta": 20},
        strategy_max_position_size={"alpha": 100, "beta": 200},
        strategy_max_order_value={"alpha": 5000, "beta": 10000},
        
        # 策略级亏损限制
        strategy_max_daily_loss={"alpha": 0.03, "beta": 0.05},
        strategy_max_drawdown={"alpha": 0.15, "beta": 0.20},
        
        # 风控后动作
        strategy_reduce_only_after_risk={"alpha": True, "beta": False},
        strategy_risk_cooldown_bars={"alpha": 2, "beta": 0},
        
        # 优先级
        strategy_priority={"alpha": 1, "beta": 2},
    ),
)

result = run_backtest(strategy=AlphaStrategy, data=df, config=config)

风控后动作

仅平仓模式：

strategy_reduce_only_after_risk={"alpha": True}

• 风控触发后，该策略只能提交平仓订单
• 开仓订单会被拒绝

冷却期：

strategy_risk_cooldown_bars={"alpha": 2}

• 风控触发后，该策略在 N 个 Bar 内无法下单
• 冷却期结束后恢复正常

---

最佳实践

1. 分层配置

# 账户级：全局底线
risk_config = RiskConfig(
    max_account_drawdown=0.20,  # 账户最大回撤 20%
    max_daily_loss=0.05,        # 单日最大亏损 5%
)

# 策略级：策略约束
strategy_config = StrategyConfig(
    strategy_max_position_size={"alpha": 100},
    strategy_max_daily_loss={"alpha": 0.03},
)

2. 监控拒单原因

result = run_backtest(...)

# 查看所有被拒订单
rejected = result.orders_df[result.orders_df["status"] == "Rejected"]
print(rejected[["symbol", "side", "quantity", "reject_reason"]])

常见拒单原因：

• Insufficient cash：资金不足
• Exceeds max position percent：超过持仓上限
• Exceeds sector concentration：超过行业集中度
• Account drawdown exceeded：触发账户回撤限制
• Daily loss limit exceeded：触发单日亏损限制
• Insufficient available position：可用持仓不足（T+1）

3. 风控触发日志

class MyStrategy(Strategy):
    def on_reject(self, order):
        self.log(f"订单被拒: {order.symbol} {order.side} 原因: {order.reject_reason}")

4. 动态调整风控

class AdaptiveRiskStrategy(Strategy):
    def on_bar(self, bar):
        # 根据市场波动动态调整风控
        volatility = self.calculate_volatility()
        
        if volatility > 0.03:
            # 高波动：收紧风控
            self.ctx.risk_manager.config.max_position_pct = 0.05
        else:
            # 低波动：放宽风控
            self.ctx.risk_manager.config.max_position_pct = 0.10

5. 回测前验证

# 在策略类中定义参数模型
from akquant import ParamModel, IntParam, FloatParam

class StrategyParams(ParamModel):
    max_position_pct: float = FloatParam(0.10, ge=0.01, le=1.0)
    max_drawdown: float = FloatParam(0.20, ge=0.05, le=0.50)

class MyStrategy(Strategy):
    PARAM_MODEL = StrategyParams
    
    def __init__(self, max_position_pct=0.10, max_drawdown=0.20):
        self.max_position_pct = max_position_pct
        self.max_drawdown = max_drawdown

---

完整示例

from akquant import Strategy, Bar, run_backtest, BacktestConfig, StrategyConfig
from akquant.config import RiskConfig
import numpy as np

class RiskManagedStrategy(Strategy):
    warmup_period = 30
    
    def __init__(self, ma_window=20, stop_loss=0.05):
        self.ma_window = ma_window
        self.stop_loss = stop_loss
        self.entry_prices = {}
    
    def on_start(self):
        self.subscribe("600000")
    
    def on_bar(self, bar):
        closes = self.get_history(self.ma_window, bar.symbol, "close")
        if len(closes) < self.ma_window:
            return
        
        ma = np.mean(closes)
        pos = self.get_position(bar.symbol)
        
        # 开仓
        if bar.close > ma * 1.02 and pos == 0:
            self.buy(bar.symbol, 100)
            self.entry_prices[bar.symbol] = bar.close
            self.log(f"开仓: {bar.symbol} @ {bar.close}")
        
        # 止损
        elif bar.symbol in self.entry_prices:
            entry = self.entry_prices[bar.symbol]
            pnl_pct = (bar.close - entry) / entry
            
            if pnl_pct < -self.stop_loss:
                self.sell(bar.symbol, pos)
                del self.entry_prices[bar.symbol]
                self.log(f"止损: {bar.symbol} @ {bar.close}, PnL: {pnl_pct:.2%}")
    
    def on_reject(self, order):
        self.log(f"订单被拒: {order.reject_reason}", level=40)

# 运行回测
config = BacktestConfig(
    strategy_config=StrategyConfig(
        initial_cash=500_000.0,
        commission_rate=0.0003,
        t_plus_one=True,
        risk=RiskConfig(
            max_position_pct=0.20,       # 单标的不超过 20%
            max_account_drawdown=0.15,   # 最大回撤 15%
            max_daily_loss=0.03,         # 单日亏损 3%
        ),
    ),
)

result = run_backtest(
    strategy=RiskManagedStrategy,
    data=df,
    symbol="600000",
    config=config,
)

# 检查拒单
rejected = result.orders_df[result.orders_df["status"] == "Rejected"]
if len(rejected) > 0:
    print("被拒订单:")
    print(rejected[["timestamp", "symbol", "side", "reject_reason"]])

策略编写范式

本文档提供 akquant 框架的策略编写范式，涵盖策略结构设计、生命周期管理、指标计算与常见陷阱。

目录

• 策略风格选择
• 策略生命周期
• 数据预热机制
• 历史数据访问
• 指标计算
• 常见策略范式
• 常见陷阱

---

策略风格选择

akquant 提供两种策略编写风格：

推荐：优先使用类风格，便于维护和扩展。

类风格示例

from akquant import Strategy, Bar
import numpy as np

class MyStrategy(Strategy):
    warmup_period = 20
    
    def __init__(self, ma_window=20):
        self.ma_window = ma_window
        self.warmup_period = ma_window + 1
    
    def on_start(self):
        self.subscribe("600000")
    
    def on_bar(self, bar: Bar):
        history = self.get_history(self.ma_window, bar.symbol, "close")
        if len(history) < self.ma_window:
            return
        
        ma = np.mean(history)
        pos = self.get_position(bar.symbol)
        
        if bar.close > ma and pos == 0:
            self.buy(bar.symbol, 100)
        elif bar.close < ma and pos > 0:
            self.sell(bar.symbol, pos)

函数风格示例

from akquant import run_backtest

def initialize(ctx):
    ctx.ma_window = 20

def on_bar(ctx, bar):
    history = ctx.get_history(ctx.ma_window, bar.symbol, "close")
    if len(history) < ctx.ma_window:
        return
    
    import numpy as np
    ma = np.mean(history)
    pos = ctx.get_position(bar.symbol)
    
    if bar.close > ma and pos == 0:
        ctx.buy(bar.symbol, 100)
    elif bar.close < ma and pos > 0:
        ctx.sell(bar.symbol, pos)

run_backtest(strategy=on_bar, initialize=initialize, data=df)

---

策略生命周期

完整生命周期流程

__init__ → on_start → [on_resume] → [on_bar / on_tick / on_timer] → on_stop
                           ↑
                     （仅热启动时）

各阶段职责

热启动适配

class WarmStartStrategy(Strategy):
    def __init__(self):
        self.sma = None
    
    def on_start(self):
        if not self.is_restored:
            # 冷启动：初始化指标
            self.sma = SMA(30)
        else:
            # 热启动：指标已从快照恢复
            self.log("Resumed from snapshot")
        
        # 必须执行：注册指标
        self.register_indicator("sma", self.sma)
        self.subscribe("AAPL")

---

数据预热机制

为什么需要预热？

计算技术指标（如 MA、RSI）需要一定长度的历史数据。预热机制确保策略在正式开始交易前，已经加载了足够的历史数据。

设置预热期

静态设置（推荐）：

class MyStrategy(Strategy):
    warmup_period = 20  # 类属性

动态设置：

class MyStrategy(Strategy):
    def __init__(self, ma_window=20):
        self.ma_window = ma_window
        self.warmup_period = ma_window + 5  # 动态计算

预热期计算规则：

• 单均线策略：warmup_period = ma_window
• 双均线策略：warmup_period = max(fast, slow)
• 带波动率策略：warmup_period = max(ma_window, volatility_window)

预热期内的行为

• 预热期内：on_bar 正常触发，但历史数据长度可能不足
• 建议：在 on_bar 开头检查 len(history) >= required_length

def on_bar(self, bar):
    history = self.get_history(20, bar.symbol, "close")
    if len(history) < 20:
        return  # 数据不足，跳过
    # ... 正常逻辑

---

历史数据访问

get_history

返回 numpy.ndarray，性能最高。

# 获取最近 20 根 Bar 的收盘价
closes = self.get_history(count=20, symbol="AAPL", field="close")
# 返回: array([close_t-19, ..., close_t-1, close_t])

# 可用字段
closes = self.get_history(20, symbol, "close")
opens = self.get_history(20, symbol, "open")
highs = self.get_history(20, symbol, "high")
lows = self.get_history(20, symbol, "low")
volumes = self.get_history(20, symbol, "volume")

get_history_df

返回 pandas.DataFrame，适合复杂分析。

df = self.get_history_df(count=20, symbol="AAPL")
# 列: timestamp, open, high, low, close, volume, symbol

# 示例：计算波动率
import numpy as np
df = self.get_history_df(20, bar.symbol)
volatility = df["close"].pct_change().std()

访问当前 Bar 数据

def on_bar(self, bar):
    # 方式 1：通过 Bar 对象
    close = bar.close
    volume = bar.volume
    
    # 方式 2：通过快捷属性
    close = self.close
    open_price = self.open
    high = self.high
    low = self.low
    volume = self.volume
    symbol = self.symbol

访问自定义因子

# 在 DataFrame 中添加因子
df["momentum"] = df["close"] / df["open"]
df["sentiment"] = sentiment_scores

# 在策略中访问
def on_bar(self, bar):
    mom = bar.extra.get("momentum", 0.0)
    sentiment = bar.extra.get("sentiment", 0.5)

---

指标计算

使用 numpy 手动计算

import numpy as np

def on_bar(self, bar):
    closes = self.get_history(20, bar.symbol, "close")
    if len(closes) < 20:
        return
    
    # 简单移动平均
    ma = np.mean(closes)
    
    # 标准差
    std = np.std(closes)
    
    # 布林带
    upper = ma + 2 * std
    lower = ma - 2 * std
    
    # RSI
    returns = np.diff(closes)
    gains = np.where(returns > 0, returns, 0)
    losses = np.where(returns < 0, -returns, 0)
    avg_gain = np.mean(gains[-14:])
    avg_loss = np.mean(losses[-14:])
    rs = avg_gain / (avg_loss + 1e-10)
    rsi = 100 - (100 / (1 + rs))

使用内置指标

from akquant import SMA, EMA, RSI

class IndicatorStrategy(Strategy):
    def __init__(self):
        self.indicator_mode = "precompute"
        self.sma20 = SMA(20)
        self.rsi14 = RSI(14)
        self.register_precomputed_indicator("sma20", self.sma20)
        self.register_precomputed_indicator("rsi14", self.rsi14)
    
    def on_bar(self, bar):
        sma_val = self.sma20.get_value(bar.symbol, bar.timestamp)
        rsi_val = self.rsi14.get_value(bar.symbol, bar.timestamp)

---

常见策略范式

趋势跟踪策略

class TrendFollowing(Strategy):
    warmup_period = 30
    
    def __init__(self, fast=10, slow=20):
        self.fast = fast
        self.slow = slow
        self.warmup_period = slow + 1
    
    def on_bar(self, bar):
        fast_ma = np.mean(self.get_history(self.fast, bar.symbol, "close"))
        slow_ma = np.mean(self.get_history(self.slow, bar.symbol, "close"))
        
        pos = self.get_position(bar.symbol)
        if fast_ma > slow_ma and pos == 0:
            self.buy(bar.symbol, 100)
        elif fast_ma < slow_ma and pos > 0:
            self.sell(bar.symbol, pos)

均值回归策略

class MeanReversion(Strategy):
    warmup_period = 20
    
    def __init__(self, window=20, z_threshold=2.0):
        self.window = window
        self.z_threshold = z_threshold
    
    def on_bar(self, bar):
        closes = self.get_history(self.window, bar.symbol, "close")
        if len(closes) < self.window:
            return
        
        mean = np.mean(closes)
        std = np.std(closes)
        z_score = (bar.close - mean) / std
        
        pos = self.get_position(bar.symbol)
        if z_score < -self.z_threshold and pos == 0:
            self.buy(bar.symbol, 100)
        elif z_score > self.z_threshold and pos > 0:
            self.sell(bar.symbol, pos)

带止损止盈的策略

class StrategyWithStopLoss(Strategy):
    warmup_period = 20
    
    def __init__(self, stop_loss=0.05, take_profit=0.10):
        self.stop_loss = stop_loss
        self.take_profit = take_profit
        self.entry_price = {}
    
    def on_bar(self, bar):
        # ... 开仓逻辑
        if buy_signal and pos == 0:
            self.buy(bar.symbol, 100)
            self.entry_price[bar.symbol] = bar.close
        
        # 止损止盈
        if bar.symbol in self.entry_price:
            entry = self.entry_price[bar.symbol]
            pnl_pct = (bar.close - entry) / entry
            
            if pnl_pct <= -self.stop_loss or pnl_pct >= self.take_profit:
                self.sell(bar.symbol, pos)
                del self.entry_price[bar.symbol]

---

常见陷阱

1. 预热期不足

# ❌ 错误：warmup_period 小于指标窗口
class BadStrategy(Strategy):
    warmup_period = 10
    
    def on_bar(self, bar):
        history = self.get_history(20, bar.symbol, "close")  # 需要 20 根
        ma = np.mean(history)  # 实际可能只有 10 根，导致错误

# ✅ 正确：warmup_period >= 最大窗口
class GoodStrategy(Strategy):
    warmup_period = 25  # 留出缓冲
    
    def on_bar(self, bar):
        history = self.get_history(20, bar.symbol, "close")
        if len(history) < 20:
            return
        ma = np.mean(history)

2. T+1 规则未区分

# ❌ 错误：T+1 模式下用总持仓判断
def on_bar(self, bar):
    pos = self.get_position(bar.symbol)  # 总持仓
    if sell_signal and pos > 0:
        self.sell(bar.symbol, pos)  # 可能卖出今日买入的部分

# ✅ 正确：T+1 模式下用可用持仓
def on_bar(self, bar):
    available = self.get_available_position(bar.symbol)  # 可卖数量
    if sell_signal and available > 0:
        self.sell(bar.symbol, available)

3. 未检查数据长度

# ❌ 错误：直接计算，可能因数据不足报错
def on_bar(self, bar):
    history = self.get_history(20, bar.symbol, "close")
    ma = np.mean(history)  # 如果 history 长度 < 20 会怎样？

# ✅ 正确：防御性检查
def on_bar(self, bar):
    history = self.get_history(20, bar.symbol, "close")
    if len(history) < 20:
        return
    ma = np.mean(history)

4. 订单未成交就重复下单

# ❌ 错误：每个 Bar 都下单，导致重复
def on_bar(self, bar):
    if buy_signal:
        self.buy(bar.symbol, 100)  # 每个 Bar 都会买

# ✅ 正确：检查持仓或挂单
def on_bar(self, bar):
    pos = self.get_position(bar.symbol)
    open_orders = self.get_open_orders(bar.symbol)
    
    if buy_signal and pos == 0 and len(open_orders) == 0:
        self.buy(bar.symbol, 100)

5. on_start 未订阅数据

# ❌ 错误：忘记订阅
class BadStrategy(Strategy):
    def on_start(self):
        pass  # 未订阅
    
    def on_bar(self, bar):
        # 永远不会触发

# ✅ 正确：显式订阅
class GoodStrategy(Strategy):
    def on_start(self):
        self.subscribe("600000")
    
    def on_bar(self, bar):
        # 正常触发

---

最佳实践清单

1. 预热期：设置为最大指标窗口 + 缓冲（如 +5） 2. 数据检查：on_bar 开头检查 len(history) >= required 3. T+1 规则：A股策略设置 t_plus_one=True，用 available_position 4. 订单管理：开仓前检查持仓和挂单，避免重复下单 5. 日志记录：关键决策点使用 self.log() 记录 6. 参数命名：使用 PARAM_MODEL 定义参数，支持页面化配置 7. 风控配置：生产策略必须配置风控规则 8. 异常处理：重写 on_error 处理回调异常