akquant

AKQuant Quantitative Strategy Development Guide

Task Objective

This Skill is used to assist AI coding agents in generating executable quantitative strategy code that conforms to the akquant framework specifications. Capabilities include strategy design, backtest configuration, order management, risk control rules, parameter optimization, and cross-sectional strategy implementation.

Core Capability List

• Strategy class generation: inherit from the Strategy base class and implement lifecycle hooks
• Data interface configuration: prepare DataFrame data, set the warm-up period, access historical data
• Event-driven mechanism: callbacks such as on_bar/on_tick/on_order/on_trade
• Order management: market orders/limit orders/target positions/OCO/Bracket/Trailing Stop
• Risk control rules: position limits/drawdown circuit breaker/stop-loss thresholds/industry concentration
• Parameter optimization: grid search and rolling optimization (Walk-Forward)
• Multi-strategy orchestration: slot mapping and strategy-level risk control

Trigger Conditions

Triggered when the user expresses the following intentions:

• Develop quantitative trading strategies
• Configure the backtesting environment and parameters
• Set risk control rules
• Perform parameter optimization and tuning
• Implement cross-sectional or rotation strategies
• Troubleshoot strategy runtime errors

Strategy Development Workflow

Phase 1: Understand Requirements

1. Identify the strategy type: trend following, mean reversion, cross-sectional rotation, arbitrage, etc.
2. Determine data requirements: time period, instrument scope, field requirements
3. Clarify risk control constraints: position limits, stop-loss and take-profit, drawdown limits

Phase 2: Design the Strategy Structure

Refer to strategy-patterns.md to choose a paradigm:

• Class style (recommended): inherit from Strategy, encapsulate state and logic
• Function style: initialize + on_bar, rapid prototyping

Key decision points:

• Warm-up period setting: calculated based on the indicator window length
• Historical data access: get_history (numpy) or get_history_df (DataFrame)
• Execution mode: NextOpen (next bar open) or CurrentClose (current bar close)

Phase 3: Write the Strategy Code

Use assets/strategy-template.py as a starting point:

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)

Phase 4: Configure the Backtesting Environment

Refer to api-reference.md to set parameters:

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

Phase 5: Set Risk Control Rules

Refer to risk-management.md to configure:

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

Phase 6: Parameter Optimization

Refer to optimization.md to execute:

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-Sectional Strategy Development

Refer to cross-section-guide.md to implement multi-instrument rotation:

Recommended paradigm: use on_timer to uniformly trigger rebalancing

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)

Resource Index

Resource	Purpose	When to Read
api-reference.md	API Quick Reference	Look up function signatures and parameters
strategy-patterns.md	Strategy Paradigms	Design the strategy structure
risk-management.md	Risk Control Configuration	Set risk control rules
optimization.md	Parameter Optimization	Tuning strategy parameters
cross-section-guide.md	Cross-sectional strategy	Implement multi-asset rotation
strategy-template.py	Strategy template	Quickly generate a code skeleton

Environment setup and dependency management

Use uv to manage the project environment (recommended)

Since akquant depends on pandas>=3.0.0, a global installation may have version conflicts with existing projects. It is recommended to use uv to create an isolated environment:

1. Install uv: skip if already installed

# 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. Create a project and initialize the environment

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

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

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

3. Run the strategy script

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

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

4. Dependency version locking

uv will automatically generate the uv.lock file to ensure consistent dependencies across the team:

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

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

5. Suggested project structure

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

Quick start commands

# 一键创建并运行策略项目
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

Notes

1. Warm-up period calculation: Ensure warmup_period >= the maximum window length required by the indicators
2. T+1 rule: A-share strategies need to set t_plus_one=True, and distinguish between total positions and available positions
3. Risk control priority: explicit parameters > configuration object > default values
4. Data format: DataFrame must contain the date/open/high/low/close/volume/symbol fields
5. Cross-sectional triggering: Prefer on_timer; if there is no fixed time point, then consider the timestamp collection scheme
6. Optimization risk: Grid search is prone to overfitting; rolling optimization is recommended to verify robustness

Usage examples

Example 1: Dual moving average strategy

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)

Example 2: Trend strategy with risk control

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

Example 3: Cross-sectional momentum rotation

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)