启睿策略回测助手

python3 scripts/init_skill.py

( command -v asdkcli >/dev/null 2>&1 || { mkdir -p "$HOME/.local/bin" && OS_TAG=$(uname -s | tr '[:upper:]' '[:lower:]') && case "$(uname -m)" in x86_64|amd64) ARCH_TAG=amd64;; arm64|aarch64) ARCH_TAG=arm64;; *) echo "unsupported ARCH"; exit 1;; esac && curl -fsSL -o "$HOME/.local/bin/asdkcli" "https://agent-identity-1302490086.cos.ap-guangzhou.myqcloud.com/cli/new/asdkcli_${OS_TAG}_${ARCH_TAG}" && chmod +x "$HOME/.local/bin/asdkcli" && export PATH="$HOME/.local/bin:$PATH" && RC_FILE="${ZSH_VERSION:+$HOME/.zshrc}" && RC_FILE="${RC_FILE:-${BASH_VERSION:+$HOME/.bashrc}}" && RC_FILE="${RC_FILE:-$HOME/.profile}" && touch "$RC_FILE" && { grep -q '# asdkcli PATH' "$RC_FILE" || printf 'n# asdkcli PATHnexport PATH="$HOME/.local/bin:$PATH"n' >> "$RC_FILE"; }; } )

if (-not (Get-Command asdkcli.exe -ErrorAction SilentlyContinue)) { $d="$env:USERPROFILE.localbin"; if (-not (Test-Path $d)) { New-Item -ItemType Directory -Path $d -Force | Out-Null }; Invoke-WebRequest -Uri "https://agent-identity-1302490086.cos.ap-guangzhou.myqcloud.com/cli/new/asdkcli_windows_amd64.exe" -OutFile "$dasdkcli.exe" -UseBasicParsing; if ($env:Path -notlike "*$d*") { $env:Path="$env:Path;$d" }; $u=[Environment]::GetEnvironmentVariable("Path","User"); if ($u -notlike "*$d*") { [Environment]::SetEnvironmentVariable("Path","$u;$d","User") } }

python3 scripts/init_skill.py --reset

data, code, msg = gid.stock_basic(market_code=['600519.SH'])
if code.name == 'SUCCESS':
    df = data  # DataFrame
else:
    print(f"查询失败: {msg}")

deepquant-backtest/
├── SKILL.md                ← 本文件（策略规范 + 使用说明）
├── scripts/
│   ├── init_skill.py       ← 初始化脚本
│   ├── run_backtest.py     ← 回测执行脚本
│   └── backtest_analyzer.py ← 回测报告分析器
├── strategies/              ← 示例策略
│   └── dual_ma_strategy.py  ← 示例：双均线策略（日线）
└── _shared/                 ← 内嵌shared模块（独立上架用）
    ├── cache_manager.py
    ├── tier_config.py
    ├── shared_resolver.py
    └── VERSION.TXT

# 下载基础日线数据（首次必须，约几百MB）
deepquant update-data -d ~/.quest --base

# 日常增量更新
deepquant update-data -d ~/.quest --smart

# 分钟线回测需单独下载分钟数据
deepquant update-data -d ~/.quest --minbar 000001.SZ --minbar 600000.SH

# 日线回测（最常用）
python3 scripts/run_backtest.py 
    --strategy strategies/dual_ma_strategy.py 
    --start 2023-01-01 --end 2023-12-31 
    --cash 1000000 --benchmark 000300.SH

# 分钟线回测
python3 scripts/run_backtest.py 
    --strategy strategies/my_strategy.py 
    --start 2023-01-01 --end 2023-06-30 
    --frequency 1m 
    --minbar 000001.SZ

# 强制重新下载数据包
python3 scripts/run_backtest.py --strategy strategies/xxx.py 
    --start 2023-01-01 --end 2023-12-31 --force-update

# 保存收益曲线图
python3 scripts/run_backtest.py --strategy strategies/xxx.py 
    --start 2023-01-01 --end 2023-12-31 --save-plot

from deepquant.quest.apis import *

import talib
import numpy as np
import pandas as pd

def init(context):
    # 1. 定义交易标的（单标的）
    context.stock = "000001.SZ"
    
    # 或多标的
    context.stocks = ["000001.SZ", "600000.SH", "600519.SH"]
    
    # 2. 更新股票池（日线回测必须）
    update_universe(context.stock)          # 单标的
    # update_universe(context.stocks)       # 多标的
    
    # 3. 定义策略参数（方便调参）
    context.SHORTPERIOD = 12
    context.LONGPERIOD  = 26
    
    # 4. 分钟/tick 回测需要订阅行情
    # subscribe(context.stock)
    
    logger.info(f"策略初始化完成")

def handle_bar(context, bar_dict):
    # 1. 获取历史数据
    prices = history_bars(context.stock, 100, '1d', 'close_price')
    # 或获取多字段
    df = history_bars(context.stock, 100, '1d', ['open_price', 'high_price', 'low_price', 'close_price', 'volume'])
    
    # 2. 计算指标（推荐用 talib）
    # ...
    
    # 3. 获取当前持仓
    cur_qty = get_position(context.stock).quantity
    
    # 4. 下单（选择合适的方式）
    order_target_percent(context.stock, 1)   # 目标仓位占比（最常用）
    order_target_value(context.stock, 50000) # 目标持仓市值
    order_shares(context.stock, 1000)        # 指定股数
    order_value(context.stock, 10000)        # 指定金额
    order_percent(context.stock, 0.3)        # 用资金的30%买入

# 股票
order_shares(stock, amount)                      # 按股数，正买负卖
order_lots(stock, amount)                        # 按手数（100股/手）
order_value(stock, cash_amount)                  # 按金额
order_percent(stock, percent)                    # 按组合价值比例
order_target_value(stock, cash_amount)           # 调至目标市值
order_target_percent(stock, percent)             # 调至目标比例（0=清仓, 1=满仓）
order_target_portfolio({"000001.SZ": 0.3, "600000.SH": 0.7})  # 批量调仓

# 期货
buy_open(contract, amount)                       # 多开
sell_close(contract, amount)                     # 平多
sell_open(contract, amount)                      # 空开
buy_close(contract, amount)                      # 平空

# 历史 K 线（最常用）
prices = history_bars(stock, count, '1d', 'close_price')          # 返回 ndarray
df     = history_bars(stock, count, '1d')                   # 返回 DataFrame

# 当前 K 线信息
bar = bar_dict[stock]
bar.open_price / bar.close_price / bar.high_price / bar.low_price / bar.volume

# 当前快照
snap = current_snapshot(stock)

# 仓位查询
pos = get_position(stock)
pos.quantity        # 持仓股数
pos.market_value    # 持仓市值
pos.trading_pnl     # 交易盈亏

# 组合信息
context.portfolio.cash            # 可用资金
context.portfolio.total_value     # 总权益
context.portfolio.portfolio_value # 组合价值

# 股票筛选
stocks = index_components("000300.SH")  # 沪深300成分股
stocks = get_industry("银行", source=12) # 行业股票列表

import talib
import numpy as np

prices = history_bars(stock, 100, '1d', 'close_price')

# 均线
sma = talib.SMA(prices, timeperiod=20)
ema = talib.EMA(prices, timeperiod=20)

# MACD
macd, signal, hist = talib.MACD(prices, 12, 26, 9)

# RSI
rsi = talib.RSI(prices, timeperiod=14)

# 布林带
upper, mid, lower = talib.BBANDS(prices, timeperiod=20)

# ATR（需要 high/low/close）
high  = history_bars(stock, 20, '1d', 'high_price')
low   = history_bars(stock, 20, '1d', 'low_price')
close = history_bars(stock, 20, '1d', 'close_price')
atr   = talib.ATR(high, low, close, timeperiod=14)

# -*- coding: utf-8 -*-
"""
策略名称：XXX 策略
策略逻辑：（简要描述）
适用频率：日线 1d / 分钟线 1m
风险提示：...
"""
from deepquant.quest.apis import *
import talib
import numpy as np


def init(context):
    context.stock = "000001.SZ"        # 交易标的
    context.PARAM1 = 20                # 参数1
    context.PARAM2 = 60                # 参数2
    update_universe(context.stock)
    logger.info(f"策略初始化完成，标的: {context.stock}")


def before_trading(context):
    # 盘前处理，不能下单
    pass


def handle_bar(context, bar_dict):
    # 获取历史数据
    prices = history_bars(context.stock, context.PARAM2 + 1, '1d', 'close_price')
    
    # 计算指标
    # ...
    
    # 获取当前持仓
    cur_qty = get_position(context.stock).quantity
    
    # 交易逻辑
    # if 买入条件:
    #     order_target_percent(context.stock, 1)
    # if 卖出条件 and cur_qty > 0:
    #     order_target_percent(context.stock, 0)


def after_trading(context):
    # 盘后处理，不能下单
    pass

# ❌ 错误：用 poll+timeout 轮询，超时就 kill
process = exec(..., background=True)
process_poll(pid, timeout=30000)  # 30秒就嫌慢
process_kill(pid)  # kill了正在写h5的进程 → 文件损坏

# ✅ 正确：一次性等待，给足超时时间
ret = subprocess.run(cmd, timeout=600, env=sub_env)  # 10分钟，足够了
# 在 exec 之前打印："⏳ 数据下载中，预计3~10分钟，请勿中断"

python3 scripts/backtest_analyzer.py reports/<策略名>/

python3 ~/.openclaw/workspace/skills/deepquant-strategy-compare/scripts/compare_strategies.py 
    --strategy-dir reports/策略A 
    --strategy-dir reports/策略B 
    --output /tmp/strategy_comparison.png

import matplotlib.pyplot as plt
plt.rcParams['font.sans-serif'] = ['Noto Sans CJK SC', 'SimHei', 'DejaVu Sans']
plt.rcParams['axes.unicode_minus'] = False
plt.rcParams['figure.dpi'] = 150

# CentOS/RHEL
yum install -y wqy-microhei-fonts wqy-zenhei-fonts

# Ubuntu/Debian
apt-get install -y fonts-wqy-microhei fonts-wqy-zenhei