Joinquant

聚宽量化（JoinQuant）

聚宽 是中国领先的在线量化交易平台，提供免费数据查询、策略回测和模拟交易。支持A股、期货、基金、指数等品种，采用事件驱动的Python策略框架。

⚠️ 需要在 https://www.joinquant.com 注册账号。策略运行在聚宽云端，也可以通过JQData在本地获取数据。

安装 (Local Data SDK)

pip install jqdatasdk

本地数据认证

import jqdatasdk as jq

# 使用聚宽账号登录（每日免费数据额度）
jq.auth('your_username', 'your_password')

# 查看剩余数据额度
print(jq.get_query_count())

策略程序结构（在线回测）

聚宽策略采用事件驱动架构，在平台网页界面编写和运行：

def initialize(context):
    """初始化函数 — 策略启动时调用一次"""
    # 设置基准指数（沪深300）
    set_benchmark('000300.XSHG')
    # 设置手续费 and slippage
    set_order_cost(OrderCost(open_tax=0, close_tax=0.001,
                              open_commission=0.0003, close_commission=0.0003,
                              min_commission=5), type='stock')
    set_slippage(FixedSlippage(0.02))
    # 设置股票池
    g.security = '000001.XSHE'

def handle_data(context, data):
    """盘中事件 — 每个交易频率触发一次"""
    security = g.security
    # 获取最近20天收盘价
    close_data = attribute_history(security, 20, '1d', ['close'])
    ma5 = close_data['close'][-5:].mean()
    ma20 = close_data['close'].mean()

    # 金叉 — buy
    if ma5 > ma20:
        order_target_value(security, context.portfolio.total_value * 0.9)
    # 死叉 — sell
    elif ma5 < ma20:
        order_target(security, 0)

股票代码格式

数据查询函数（JQData）

行情数据

import jqdatasdk as jq

# 获取日K线数据
df = jq.get_price(
    '000001.XSHE',              # 股票代码
    start_date='2024-01-01',    # 开始日期
    end_date='2024-06-30',      # 结束日期
    frequency='daily',          # 频率: daily(日), minute(分钟), 1m, 5m, 15m, 30m, 60m, 120m
    fields=['open', 'close', 'high', 'low', 'volume', 'money'],
    skip_paused=True,           # 跳过停牌日
    fq='pre',                   # 复权: None(不复权), 'pre'(前复权), 'post'(后复权)
    panel=False                 # False返回DataFrame
)

# 获取多只股票数据
df = jq.get_price(['000001.XSHE', '600000.XSHG'],
                   start_date='2024-01-01', end_date='2024-06-30',
                   frequency='daily', fields=['close'], panel=False)

# 获取分钟级数据
df = jq.get_price('000001.XSHE', start_date='2024-06-01 09:30:00',
                   end_date='2024-06-01 15:00:00', frequency='1m')

获取最近N条数据

# 获取最近20个交易日的收盘价
df = jq.get_bars('000001.XSHE', count=20, unit='1d', fields=['close', 'volume'])

财务数据

# 查询财务指标
df = jq.get_fundamentals(
    jq.query(
        jq.valuation.code,
        jq.valuation.market_cap,          # 总市值（亿元）
        jq.valuation.pe_ratio,            # 市盈率
        jq.valuation.pb_ratio,            # 市净率
        jq.valuation.turnover_ratio,      # 换手率
        jq.indicator.roe,                 # 净资产收益率（ROE）
        jq.indicator.eps,                 # 每股收益
        jq.indicator.revenue,             # 营业收入
        jq.indicator.net_profit,          # 净利润
    ).filter(
        jq.valuation.pe_ratio > 0,        # 排除亏损股
        jq.valuation.pe_ratio < 30,       # PE小于30
        jq.valuation.market_cap > 100     # 市值大于100亿元
    ).order_by(
        jq.valuation.market_cap.desc()    # 按市值降序排列
    ).limit(50),                          # 取前50
    date='2024-06-30'
)
print(df)

指数成分股

# 获取沪深300成分股
stocks = jq.get_index_stocks('000300.XSHG')
print(f'沪深300共 {len(stocks)} 只成分股')

# 获取行业成分股
stocks = jq.get_industry_stocks('I64')  # 银行业

行业分类

# 获取股票所属行业
industry = jq.get_industry('000001.XSHE')
print(industry)

# 获取申万一级行业列表
industries = jq.get_industries(name='sw_l1')
print(industries)

交易日历

# 获取交易日列表
days = jq.get_trade_days(start_date='2024-01-01', end_date='2024-06-30')

# 获取全部交易日
all_days = jq.get_all_trade_days()

股票基本信息

# 获取全部A股上市公司
stocks = jq.get_all_securities(types=['stock'], date='2024-06-30')
print(f'A股总数: {len(stocks)}')

# 获取单只股票信息
info = jq.get_security_info('000001.XSHE')
print(f'名称: {info.display_name}, 上市日期: {info.start_date}')

# 获取ST股票
st_stocks = jq.get_extras('is_st', ['000001.XSHE'], start_date='2024-01-01', end_date='2024-06-30')

龙虎榜数据

# 获取龙虎榜数据
df = jq.get_billboard_list(stock_list=None, start_date='2024-06-01', end_date='2024-06-30')
print(df.head())

融资融券数据

# 获取融资融券汇总数据
df = jq.get_mtss('000001.XSHE', start_date='2024-01-01', end_date='2024-06-30')
print(df.head())

交易函数（在线策略）

按数量下单

# 买入100股
order('000001.XSHE', 100)

# 卖出200股
order('000001.XSHE', -200)

# 限价买入
order('000001.XSHE', 100, LimitOrderStyle(11.50))

# 市价买入
order('000001.XSHE', 100, MarketOrderStyle())

调仓到目标

# 调仓到目标数量
order_target('000001.XSHE', 1000)     # Adjust to hold 1000 shares
order_target('000001.XSHE', 0)        # Liquidate position

# 调仓到目标金额
order_target_value('000001.XSHE', 100000)  # Adjust to 100,000 CNY market value

# 调仓到目标比例（占总资产）
order_target_percent('000001.XSHE', 0.3)   # Adjust to 30% of total assets

撤单

# 获取未成交订单
open_orders = get_open_orders()
# 撤销指定订单
cancel_order(order_id)

账户与持仓查询

def handle_data(context, data):
    # 账户信息rmation
    cash = context.portfolio.available_cash       # 可用资金
    total = context.portfolio.total_value          # 总资产
    positions_value = context.portfolio.positions_value  # 持仓市值

    # 查询持仓
    for stock, pos in context.portfolio.positions.items():
        print(f'{stock}: Quantity={pos.total_amount}, '
              f'Sellable={pos.closeable_amount}, '
              f'Cost={pos.avg_cost:.2f}, '
              f'Current price={pos.price:.2f}, '
              f'Market value={pos.value:.2f}')

定时任务

def initialize(context):
    # 每日盘前执行
    run_daily(before_market_open, time='before_open')
    # 每日指定时间执行
    run_daily(market_open, time='09:35')
    run_daily(afternoon_check, time='14:50')
    # 每周执行
    run_weekly(weekly_rebalance, weekday=1, time='09:35')  # 每周一
    # 每月执行
    run_monthly(monthly_rebalance, monthday=1, time='09:35')  # 每月1号

def before_market_open(context):
    log.info('Pre-market preparation')

def market_open(context):
    log.info('Market open trading')

风险管理

def initialize(context):
    g.security = '000001.XSHE'
    set_benchmark('000300.XSHG')

def handle_data(context, data):
    security = g.security
    # 获取当前价格
    current_price = data[security].close

    # 检查持仓盈亏
    if security in context.portfolio.positions:
        pos = context.portfolio.positions[security]
        cost = pos.avg_cost
        pnl_ratio = (current_price - cost) / cost

        # 盈利10%止盈
        if pnl_ratio >= 0.10:
            order_target(security, 0)
            log.info(f'Take profit: {security} gain {pnl_ratio:.2%}')
        # 亏损5%止损
        elif pnl_ratio <= -0.05:
            order_target(security, 0)
            log.info(f'Stop loss: {security} loss {pnl_ratio:.2%}')

完整示例 — Multi-Factor Stock Selection Strategy

def initialize(context):
    set_benchmark('000300.XSHG')
    set_order_cost(OrderCost(open_tax=0, close_tax=0.001,
                              open_commission=0.0003, close_commission=0.0003,
                              min_commission=5), type='stock')
    g.hold_num = 10  # 持股数量
    # 每月第一个交易日调仓
    run_monthly(rebalance, monthday=1, time='09:35')

def rebalance(context):
    # 多因子选股
    df = get_fundamentals(
        query(
            valuation.code,
            valuation.pe_ratio,
            valuation.pb_ratio,
            valuation.market_cap,
            indicator.roe,
        ).filter(
            valuation.pe_ratio > 5,
            valuation.pe_ratio < 25,
            valuation.pb_ratio > 0.5,
            valuation.pb_ratio < 5,
            indicator.roe > 10,
            valuation.market_cap > 100,
        ).order_by(
            valuation.pe_ratio.asc()
        ).limit(g.hold_num)
    )

    target_stocks = list(df['code'])
    log.info(f'Selected {len(target_stocks)} stocks')

    # 卖出不在目标列表中的股票
    for stock in context.portfolio.positions:
        if stock not in target_stocks:
            order_target(stock, 0)

    # 等权重买入目标股票
    if target_stocks:
        per_value = context.portfolio.total_value * 0.95 / len(target_stocks)
        for stock in target_stocks:
            order_target_value(stock, per_value)

def handle_data(context, data):
    pass

进阶示例

行业轮动策略

def initialize(context):
    set_benchmark('000300.XSHG')
    g.hold_num = 5
    g.industry_etfs = {
        '512010.XSHG': '医药ETF',
        '512880.XSHG': '证券ETF',
        '512800.XSHG': '银行ETF',
        '515030.XSHG': '新能源车ETF',
        '159995.XSHE': '芯片ETF',
        '512690.XSHG': '白酒ETF',
        '510300.XSHG': '沪深300ETF',
        '159915.XSHE': '创业板ETF',
    }
    run_monthly(rebalance, monthday=1, time='09:35')

def rebalance(context):
    """按20日动量排序，选取最强的前N只ETF"""
    momentum = {}
    for etf, name in g.industry_etfs.items():
        df = attribute_history(etf, 20, '1d', ['close'])
        if len(df) >= 20:
            ret = (df['close'].iloc[-1] / df['close'].iloc[0]) - 1
            momentum[etf] = ret

    # 按动量排序
    sorted_etfs = sorted(momentum.items(), key=lambda x: x[1], reverse=True)
    targets = [etf for etf, _ in sorted_etfs[:g.hold_num]]
    log.info(f'This month selected: {[g.industry_etfs[e] for e in targets]}')

    # 卖出不在目标列表中的持仓
    for stock in context.portfolio.positions:
        if stock not in targets:
            order_target(stock, 0)

    # 等权重买入
    per_value = context.portfolio.total_value * 0.95 / len(targets)
    for etf in targets:
        order_target_value(etf, per_value)

def handle_data(context, data):
    pass

数据研究 — 全市场财务筛选

import jqdatasdk as jq
import pandas as pd

jq.auth('your_username', 'your_password')

# 全市场财务筛选：低PE + 高ROE + 高成长
df = jq.get_fundamentals(
    jq.query(
        jq.valuation.code,
        jq.valuation.pe_ratio,
        jq.valuation.pb_ratio,
        jq.valuation.market_cap,
        jq.indicator.roe,
        jq.indicator.inc_revenue_year_on_year,    # 营收同比增长率
        jq.indicator.inc_net_profit_year_on_year,  # 净利润同比增长率
    ).filter(
        jq.valuation.pe_ratio > 5,
        jq.valuation.pe_ratio < 20,
        jq.indicator.roe > 15,
        jq.indicator.inc_revenue_year_on_year > 10,
        jq.indicator.inc_net_profit_year_on_year > 10,
        jq.valuation.market_cap > 50,
    ).order_by(
        jq.indicator.roe.desc()
    ).limit(30),
    date='2024-06-30'
)

print(f'筛选出 {len(df)} 只股票:')
print(df.to_string(index=False))

# 导出为CSV
df.to_csv('selected_stocks.csv', index=False)

使用技巧

• 聚宽策略运行在云端 — 无需本地安装交易环境。
• JQData本地SDK有免费每日数据额度，适合数据研究。
• 在策略中使用 attribute_history 获取历史数据；在研究环境中使用 get_price。
• 回测时使用 set_order_cost 设置真实手续费 — 默认手续费为0。
• g 对象用于在函数之间持久化变量（类似Ptrade）。
• 文档：https://www.joinquant.com/help/api/help

🤖 AI Agent 高阶使用指南

对于 AI Agent，在使用该量化/数据工具时应遵循以下高阶策略和最佳实践，以确保任务的高效完成：

1. 数据校验与错误处理

在获取数据或执行操作后，AI 应当主动检查返回的结果格式是否符合预期，以及是否存在缺失值（NaN）或空数据。

• 示例策略：在通过 API 获取数据框（DataFrame）后，使用 if df.empty: 进行校验；捕获 Exception 以防网络或接口错误导致进程崩溃。

2. 多步组合分析

AI 经常需要进行宏观经济分析或跨市场对比。应善于将当前接口与其他数据源或工具组合使用。

• 示例策略：先获取板块或指数的宏观数据，再筛选成分股，最后对具体标的进行深入的财务或技术面分析，形成完整的决策链条。

3. 构建动态监控与日志

对于交易和策略类任务，AI 可以定期拉取数据并建立监控机制。

• 示例策略：使用循环或定时任务检查特定标的的异动（如涨跌停、放量），并在发现满足条件的信号时输出结构化日志或触发预警。

社区与支持

由 大佬量化 维护 — 量化交易教学与策略研发团队。

微信客服: bossquant1 · Bilibili · 搜索 大佬量化 — 微信公众号 / Bilibili / 抖音