akquant

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

AKQuant 量化策略开发指南

任务目标

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

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

核心能力清单

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

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

触发条件

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

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

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

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

策略开发工作流

阶段一：理解需求

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

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

阶段二：设计策略结构

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

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

关键决策点：

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

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

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

关键决策点：

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

阶段三：编写策略代码

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

python

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)

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

python

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 设置参数：

python

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",
)

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

python

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 配置：

python

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%
    ),
)

参考 risk-management.md 配置：

python

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 执行：

python

from akquant import run_grid_search, run_walk_forward

参考 optimization.md 执行：

python

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", )

undefined

wfo_results = run_walk_forward( strategy=MyStrategy, param_grid={"param1": [10, 20, 30]}, data=df, train_period=250, test_period=60, metric="sharpe_ratio", )

undefined

横截面策略开发

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

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

python

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)

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

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

python

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)

资源索引

资源	用途	何时读取
api-reference.md	API 速查	查询函数签名与参数
strategy-patterns.md	策略范式	设计策略结构
risk-management.md	风控配置	设置风控规则
optimization.md	参数优化	调优策略参数
cross-section-guide.md	横截面策略	实现多标的轮动
strategy-template.py	策略模板	快速生成代码骨架

资源	用途	何时读取
api-reference.md	API 速查	查询函数签名与参数
strategy-patterns.md	策略范式	设计策略结构
risk-management.md	风控配置	设置风控规则
optimization.md	参数优化	调优策略参数
cross-section-guide.md	横截面策略	实现多标的轮动
strategy-template.py	策略模板	快速生成代码骨架

环境准备与依赖管理

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

由于 akquant 依赖

pandas>=3.0.0

，全局安装可能与现有项目存在版本冲突。推荐使用 uv 创建隔离环境：

由于 akquant 依赖

pandas>=3.0.0

，全局安装可能与现有项目存在版本冲突。推荐使用 uv 创建隔离环境：

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

bash

undefined

bash

undefined

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"

undefined

powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"

undefined

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

bash

undefined

bash

undefined

创建项目目录

mkdir my-quant-strategy cd my-quant-strategy

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

uv init

创建虚拟环境并安装依赖

uv venv uv add akquant pandas numpy

undefined

uv venv uv add akquant pandas numpy

undefined

3. 运行策略脚本

bash

undefined

bash

undefined

方式一：使用 uv run（推荐）

uv run python my_strategy.py

方式二：激活虚拟环境后运行

source .venv/bin/activate # macOS/Linux

.venv\Scripts\activate # Windows

python my_strategy.py

undefined

python my_strategy.py

undefined

4. 依赖版本锁定

uv 会自动生成

uv.lock

文件，确保团队依赖一致：

bash

undefined

uv 会自动生成

uv.lock

文件，确保团队依赖一致：

bash

undefined

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

uv sync

添加新依赖

uv add scipy # 自动更新 lock 文件

undefined

uv add scipy # 自动更新 lock 文件

undefined

5. 项目结构建议

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

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

快速启动命令

bash

undefined

bash

undefined

一键创建并运行策略项目

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)

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

undefined

uv run python strategy.py

undefined

注意事项

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

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

使用示例

示例 1：双均线策略

python

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)

python

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：带风控的趋势策略

python

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)

python

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, ), )

undefined

undefined

示例 3：横截面动量轮动

python

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)

python

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)