掘金量化（MyQuant / GoldMiner）

掘金量化 是专业量化交易平台，提供事件驱动策略开发、回测、模拟交易和实盘交易。支持A股、期货、期权、ETF、可转债和融资融券，通过统一的Python SDK（gm.api）操作。

需要在 https://www.myquant.cn 注册获取Token进行认证。实盘交易需要通过掘金终端连接券商账户。

安装

pip install gm

也可从官网下载：https://www.myquant.cn/docs/downloads/698

架构概述

你的Python脚本（strategy.py）
    ↓ gm SDK（from gm.api import *）
    ├── 回测模式  → 掘金回测引擎（云端/本地）
    ├── 模拟交易   → 掘金模拟交易服务器
    └── 实盘交易    → 掘金终端 → 券商网关

策略程序结构

所有策略遵循事件驱动架构，包含生命周期函数：

from gm.api import *

def init(context):
    """Initialization function — called once when the strategy starts, used to set up subscriptions and parameters"""
    # Subscribe to 30-second bars for two stocks, keeping the most recent 5 historical bars
    subscribe(symbols='SHSE.600000,SZSE.000001', frequency='30s', count=5)
    context.my_param = 0.8  # Custom context attribute

def on_bar(context, bars):
    """Bar event — triggered at each bar close, write main trading logic here"""
    bar = bars[0]
    if bar['close'] > bar['open']:
        # Close price above open price, place a limit buy order for 1000 shares
        order_volume(symbol=bar['symbol'], volume=1000,
                     side=OrderSide_Buy, order_type=OrderType_Limit,
                     position_effect=PositionEffect_Open, price=bar['close'])

def on_tick(context, tick):
    """Tick event — triggered on each tick push (real-time mode)"""
    print(tick['symbol'], tick['price'])

def on_order_status(context, order):
    """Order status event — triggered when order status changes"""
    print(f"Order {order['cl_ord_id']}: status={order['status']}")

def on_execution_report(context, execrpt):
    """Execution report event — triggered on trade execution"""
    print(f"Executed: {execrpt['symbol']} {execrpt['filled_volume']}@{execrpt['filled_vwap']}")

if __name__ == '__main__':
    run(strategy_id='your_strategy_id',
        filename='main.py',
        mode=MODE_BACKTEST,                          # Run mode: backtest
        token='your_token',
        backtest_start_time='2024-01-01 09:30:00',   # Backtest start time
        backtest_end_time='2024-06-30 15:00:00',     # Backtest end time
        backtest_initial_cash=1000000,                # Initial capital 1,000,000
        backtest_commission_ratio=0.0003,             # Commission rate 0.03%
        backtest_slippage_ratio=0.001,                # Slippage 0.1%
        backtest_adjust=ADJUST_PREV)                  # 前复权

基本函数

init — 策略初始化

def init(context):
    # Subscribe to daily bars, keeping the most recent 20
    subscribe(symbols='SHSE.600000', frequency='1d', count=20)
    context.ratio = 0.8  # Custom parameter

• 策略启动时调用一次
• 用于设置订阅、定义参数和初始化状态
• 回测模式下init中不允许下单（模拟/实盘模式允许）

schedule — 定时任务

def init(context):
    # Execute algo_1 function at 09:40:00 on each trading day
    schedule(schedule_func=algo_1, date_rule='1d', time_rule='09:40:00')
    # Execute algo_2 function at 14:30:00 on the 1st trading day of each month
    schedule(schedule_func=algo_2, date_rule='1m', time_rule='14:30:00')

def algo_1(context):
    # Stock selection / rebalancing logic
    pass

def algo_2(context):
    # Market order to buy 200 shares
    order_volume(symbol='SHSE.600000', volume=200,
                 side=OrderSide_Buy, order_type=OrderType_Market,
                 position_effect=PositionEffect_Open)

• date_rule: '1d' (daily), '1w' (weekly, backtest only), '1m' (monthly, backtest only)
• time_rule: 'HH:MM:SS' format, must be zero-padded (e.g., '09:40:00', not '9:40:0')

run — 运行策略

run(strategy_id='strategy_1',
    filename='main.py',
    mode=MODE_BACKTEST,       # MODE_BACKTEST=2 (backtest), MODE_LIVE=1 (live/paper)
    token='your_token',
    backtest_start_time='2024-01-01 09:30:00',
    backtest_end_time='2024-06-30 15:00:00',
    backtest_initial_cash=1000000,          # Initial capital
    backtest_transaction_ratio=1,           # Transaction fill ratio
    backtest_commission_ratio=0.0003,       # Commission rate
    backtest_slippage_ratio=0.001,          # Slippage ratio
    backtest_adjust=ADJUST_PREV)  # ADJUST_NONE=0 (unadjusted), ADJUST_PREV=1 (forward-adjusted), ADJUST_POST=2 (backward-adjusted)

stop — 停止策略

stop()  # Gracefully stop and exit the strategy

数据订阅

subscribe — 订阅行情数据

subscribe(symbols='SHSE.600000,SZSE.000001',
          frequency='60s',    # Frequency: tick, 1s~300s, 60s, 300s, 900s, 1800s, 3600s, 1d
   
          count=5,            # Number of historical bars to keep in context.data
          wait_group=True,    # Wait for all symbols in the group before triggering callback
          wait_group_timeout='5s')  # Wait timeout duration

unsubscribe — 取消订阅

unsubscribe(symbols='SHSE.600000', frequency='60s')

数据事件

数据查询函数

current — 实时快照

data = current(symbols='SZSE.000001')
# 收益率: [{'symbol': 'SZSE.000001', 'price': 16.56, 'open': 16.20, 'high': 16.92, 'low': 16.15,
#            'quotes': [{'bid_p': 16.55, 'bid_v': 209200, 'ask_p': 16.56, 'ask_v': 296455}, ...],
#            'cum_volume': 160006232, 'cum_amount': 2654379585.66, 'created_at': ...}]

• 回测模式：仅返回 symbol、price、created_at
• 实盘模式：返回完整tick字段，包含5档买卖盘

history — 历史K线/Tick

df = history(symbol='SHSE.000300', frequency='1d',
             start_time='2024-01-01', end_time='2024-06-30',
             fields='open,close,high,low,volume,eob',
             skip_suspended=True,          # 跳过停牌日
             fill_missing=None,            # Missing value fill: None, 'NaN', 'Last'
             adjust=ADJUST_PREV,           # 0=unadjusted, 1=forward-adjusted, 2=backward-adjusted
             df=True)                      # True returns DataFrame, False returns list[dict]

history_n — 最近N根K线

df = history_n(symbol='SHSE.600000', frequency='1d', count=20,
               end_time='2024-06-30', fields='close,volume',
               adjust=ADJUST_PREV, df=True)

context.data — 订阅数据缓存

# Access the subscribed data buffer (set by the count parameter in subscribe)
bars = context.data(symbol='SHSE.600000', frequency='60s', count=10)

Level 2历史数据

get_history_l2ticks(symbol, start_time, end_time, fields, skip_suspended, fill_missing, adjust, df)
get_history_l2bars(symbol, frequency, start_time, end_time, fields, skip_suspended, fill_missing, adjust, df)
get_history_l2transactions(symbol, start_time, end_time, fields, df)
get_history_l2orders(symbol, start_time, end_time, fields, df)
get_history_l2orders_queue(symbol, start_time, end_time, fields, df)

基本面数据

# Retrieve financial indicator data with filtering and sorting support
df = get_fundamentals(
    table='trading_derivative_indicator',    # Data table name
    symbols='SHSE.600000,SZSE.000001',       # Stock symbols
    start_date='2024-01-01',
    end_date='2024-06-30',
    fields='TCLOSE,NEGOTIABLEMV,TOTMKTCAP,TURNRATE,PETTM',  # Field list
    limit=1000,                              # Result count limit
    df=True
)

# Retrieve the most recent N records
df = get_fundamentals_n(
    table='trading_derivative_indicator',
    symbols='SHSE.600000',
    end_date='2024-06-30',
    count=10,
    fields='TCLOSE,PETTM',
    df=True
)

可用的基本面数据表：参见财务数据文档

合约信息

# Get all A-share instruments on the Shenzhen exchange
instruments = get_instruments(exchanges='SZSE', sec_types=1, df=True)

# Get information for specific instruments
info = get_instruments(symbols='SHSE.600000,SZSE.000001', df=True)
# Fields: symbol, sec_name, exchange, listed_date, delisted_date, sec_type,
#         pre_close, upper_limit, lower_limit, adj_factor, is_st, is_suspended, ...

# Get historical instrument information
hist_info = get_history_instruments(symbols='SHSE.600000', start_date='2024-01-01', end_date='2024-06-30')

# Get basic instrument information (static)
basic_info = get_instrumentinfos(symbols='SHSE.600000', df=True)

指数成分股

# Get current constituents
stocks = get_constituents(index='SHSE.000300')  # CSI 300

# Get historical constituents
hist = get_history_constituents(index='SHSE.000300', start_date='2024-01-01', end_date='2024-06-30')

行业成分股

# Get the stock list for a specified industry code
stocks = get_industry(code='J66')  # 收益率 stocks in industry J66 (Banking)

分红数据

divs = get_dividend(symbol='SHSE.600000', start_date='2020-01-01', end_date='2024-12-31', df=True)

交易日历

dates = get_trading_dates(exchange='SZSE', start_date='2024-01-01', end_date='2024-06-30')
prev = get_previous_trading_date(exchange='SHSE', date='2024-03-15')   # Previous trading date
next_d = get_next_trading_date(exchange='SHSE', date='2024-03-15')     # Next trading date

连续合约（期货）

contracts = get_continuous_contracts(csymbol='CFFEX.IF', start_date='2024-01-01', end_date='2024-06-30')

新股申购函数

quota = ipo_get_quota()                  # Query IPO subscription quota
instruments = ipo_get_instruments()      # Query today's IPO list
match_nums = ipo_get_match_number()      # Query allotment numbers
lot_info = ipo_get_lot_info()            # Query winning lot information

交易函数

按数量下单

order = order_volume(
    symbol='SHSE.600000',
    volume=10000,                               # Order quantity
    side=OrderSide_Buy,                         # Side: OrderSide_Buy=1 (buy), OrderSide_Sell=2 (sell)
    order_type=OrderType_Limit,                 # Order type: OrderType_Limit=1 (limit), OrderType_Market=2 (market)
    position_effect=PositionEffect_Open,        # Position effect: Open=1 (open), Close=2 (close), CloseToday=3 (close today), CloseYesterday=4 (close yesterday)
    price=11.0,                                 # Order price
    account=''                                  # Optional: specify account for multi-account setups
)

按金额下单

order = order_value(
    symbol='SHSE.600000',
    value=100000,                   # Target order value (CNY)
    side=OrderSide_Buy,
    order_type=OrderType_Limit,
    position_effect=PositionEffect_Open,
    price=11.0
)
# Actual quantity = value / price, truncated to valid lot size

按比例下单

order = order_percent(
    symbol='SHSE.600000',
    percent=0.1,                    # 10% of total assets
    side=OrderSide_Buy,
    order_type=OrderType_Limit,
    position_effect=PositionEffect_Open,
    price=11.0
)

目标仓位函数

# Adjust to target volume
order_target_volume(symbol='SHSE.600000', volume=10000,
                    position_side=PositionSide_Long,
                    order_type=OrderType_Limit, price=13.0)

# Adjust to target value
order_target_value(symbol='SHSE.600000', value=130000,
                   position_side=PositionSide_Long,
                   order_type=OrderType_Limit, price=13.0)

# Adjust to target percentage (percentage of total assets)
order_target_percent(symbol='SHSE.600000', percent=0.1,
                     position_side=PositionSide_Long,
                     order_type=OrderType_Limit, price=13.0)

批量下单

orders = [
    {'symbol': 'SHSE.600000', 'volume': 1000, 'side': OrderSide_Buy,
     'order_type': OrderType_Limit, 'position_effect': PositionEffect_Open, 'price': 11.0},
    {'symbol': 'SZSE.000001', 'volume': 2000, 'side': OrderSide_Buy,
     'order_type': OrderType_Limit, 'position_effect': PositionEffect_Open, 'price': 15.0},
]
results = order_batch(orders)  # Submit orders in batch

撤单与全部平仓

# 撤销指定订单
order_cancel(wait_cancel_orders=[
    {'cl_ord_id': order1['cl_ord_id'], 'account_id': order1['account_id']}
])
order_cancel_all()        # Cancel all unfilled orders
order_close_all()         # Close all positions (limit orders)

查询委托

unfinished = get_unfinished_orders()    # Query unfinished orders (pending/partially filled)
all_orders = get_orders()               # Query all orders today
executions = get_execution_reports()    # Query all execution reports today

特殊交易

ipo_buy(symbol)                                       # IPO subscription
fund_etf_buy(symbol, volume, price)                   # ETF purchase
fund_etf_redemption(symbol, volume, price)             # ETF redemption
fund_subscribing(symbol, volume, price)                # Fund initial subscription
fund_buy(symbol, volume, price)                        # Fund purchase
fund_redemption(symbol, volume, price)                 # Fund redemption
bond_reverse_repurchase_agreement(symbol, volume, price) # Treasury reverse repo

账户与持仓查询

# Query cash information
cash = context.account().cash
# Attributes: nav (net asset value), fpnl (floating P&L), pnl (realized P&L), available (available funds),
#             order_frozen (order frozen), balance (balance), market_value (position market value), cum_trade (cumulative trade amount), ...

# Query all positions
positions = context.account().positions()
# Each position contains: symbol, side, volume, available_now, cost, vwap, market_value, fpnl, ...

# Query a specific position
pos = context.account().position(symbol='SHSE.600000', side=PositionSide_Long)

融资融券

# Margin buy
credit_buying_on_margin(symbol, volume, price, side=OrderSide_Buy, order_type=OrderType_Limit)

# Short sell
credit_short_selling(symbol, volume, price, side=OrderSide_Sell, order_type=OrderType_Limit)

# Repayment
credit_repay_cash_directly(amount)              # Direct cash repayment
credit_repay_share_directly(symbol, volume)      # Direct share repayment
credit_repay_share_by_buying_share(symbol, volume, price)  # Buy shares to repay
credit_repay_cash_by_selling_share(symbol, volume, price)  # Sell shares to repay cash

# Collateral operations
credit_buying_on_collateral(symbol, volume, price)   # Buy collateral
credit_selling_on_collateral(symbol, volume, price)   # Sell collateral
credit_collateral_in(symbol, volume)                  # Transfer collateral in
credit_collateral_out(symbol, volume)                 # Transfer collateral out

# Queries
credit_get_collateral_instruments()          # Query collateral instruments
credit_get_borrowable_instruments()          # Query borrowable instruments
credit_get_borrowable_instruments_positions() # Query broker's lending positions
credit_get_contracts()                       # Query margin trading contracts
credit_get_cash()                            # Query margin trading funds

算法交易

# Submit an algorithmic order (e.g., TWAP, VWAP)
algo_order(symbol, volume, side, order_type, position_effect, price, algo_name, algo_param)

# Cancel an algorithmic order
algo_order_cancel(cl_ord_id, account_id)

# Query algorithmic orders
get_algo_orders()

# Pause/resume an algorithmic order
algo_order_pause(cl_ord_id, account_id, status)  # AlgoOrderStatus_Pause / AlgoOrderStatus_Resume

# Query child orders of an algorithmic order
get_algo_child_orders(cl_ord_id, account_id)

# Algorithmic order status event
def on_algo_order_status(context, order):
    print(f"Algo order {order['cl_ord_id']}: status={order['status']}")

交易事件

动态参数

def init(context):
    # Add a parameter that can be dynamically modified in the terminal UI
    add_parameter(key='threshold', value=0.05, min=0.01, max=0.2, name='买入阈值', intro='触发买入的比例')

def on_parameter(context, parameter):
    """Triggered when a parameter is modified in the terminal UI"""
    print(f"Parameter changed: {parameter['key']} = {parameter['value']}")

# Modify parameter at runtime
set_parameter(key='threshold', value=0.08)

# Read all parameters
params = context.parameters

连接事件

代码格式

格式：交易所代码.证券代码

枚举常量参考

OrderSide — 委托方向

OrderType — 委托类型

PositionEffect — 开平标志

PositionSide — 持仓方向

OrderStatus — 订单状态

ADJUST — 复权方式

完整示例 — 双均线回测策略

from gm.api import *

def init(context):
    # Subscribe to daily bars, keeping the most recent 21
    subscribe(symbols='SHSE.600000', frequency='1d', count=21)
    context.symbol = 'SHSE.600000'

def on_bar(context, bars):
    # Get the most recent 20 daily bars
    hist = context.data(symbol=context.symbol, frequency='1d', count=20)
    closes = [bar['close'] for bar in hist]

    if len(closes) < 20:
        return  # Insufficient data, skip

    # Calculate 5-day and 20-day moving averages
    ma5 = sum(closes[-5:]) / 5
    ma20 = sum(closes) / 20
    price = bars[0]['close']

    # Query current position
    pos = context.account().position(symbol=context.symbol, side=PositionSide_Long)

    # 金叉 signal: 5-day MA > 20-day MA and no position, buy
    if ma5 > ma20 and (pos is None or pos['volume'] == 0):
        order_percent(symbol=context.symbol, percent=0.9,
                      side=OrderSide_Buy, order_type=OrderType_Limit,
                      position_effect=PositionEffect_Open, price=price)
    # 死叉 signal: 5-day MA < 20-day MA and has position, close all
    elif ma5 < ma20 and pos is not None and pos['volume'] > 0:
        order_close_all()

def on_order_status(context, order):
    if order['status'] == OrderStatus_Filled:
        print(f"Executed: {order['symbol']} volume={order['filled_volume']} avg_price={order['filled_vwap']}")

def on_backtest_finished(context, indicator):
    print(f"Backtest completed: return={indicator['pnl_ratio']:.2%}, "
          f"Sharpe ratio={indicator['sharpe_ratio']:.2f}, "
          f"max drawdown={indicator['max_drawdown']:.2%}")

if __name__ == '__main__':
    run(strategy_id='ma_cross',
        filename='main.py',
        mode=MODE_BACKTEST,
        token='your_token',
        backtest_start_time='2023-01-01 09:30:00',
        backtest_end_time='2024-06-30 15:00:00',
        backtest_initial_cash=1000000,
        backtest_commission_ratio=0.0003,
        backtest_slippage_ratio=0.001,
        backtest_adjust=ADJUST_PREV)

完整示例 — 定时调仓策略

from gm.api import *

def init(context):
    # Rebalance at 09:35 on each trading day
    schedule(schedule_func=rebalance, date_rule='1d', time_rule='09:35:00')
    context.target_stocks = ['SHSE.600000', 'SZSE.000001', 'SHSE.601318']

def rebalance(context):
    # Equal-weight allocation, each stock gets 1/N of total assets
    target_pct = 1.0 / len(context.target_stocks)
    for sym in context.target_stocks:
        order_target_percent(symbol=sym, percent=target_pct * 0.95,
                             position_side=PositionSide_Long,
                             order_type=OrderType_Market)

if __name__ == '__main__':
    run(strategy_id='equal_weight',
        filename='main.py',
        mode=MODE_BACKTEST,
        token='your_token',
        backtest_start_time='2024-01-01 09:30:00',
        backtest_end_time='2024-06-30 15:00:00',
        backtest_initial_cash=1000000,
        backtest_adjust=ADJUST_PREV)

完整示例 — 纯数据研究（无交易）

from gm.api import *

# Set Token (can query data without starting a strategy)
set_token('your_token')

# Get historical bar data
df = history(symbol='SHSE.000300', frequency='1d',
             start_time='2024-01-01', end_time='2024-06-30',
             fields='open,close,high,low,volume',
             adjust=ADJUST_PREV, df=True)
print(df.head())

# Get financial indicator data
fund = get_fundamentals(table='trading_derivative_indicator',
                        symbols='SHSE.600000',
                        start_date='2024-01-01',
                        end_date='2024-06-30',
                        fields='TCLOSE,PETTM,TURNRATE',
                        df=True)
print(fund)

# 获取沪深300成分股
constituents = get_constituents(index='SHSE.000300')
print(f'CSI 300 has {len(constituents)} constituent stocks')

# Get banking industry stocks
bank_stocks = get_industry(code='J66')
print(f'Banking industry has {len(bank_stocks)} stocks')

支持的数据频率

运行模式

使用技巧

• 需要Token：在 https://www.myquant.cn 注册获取。使用 set_token() 或在 run() 中传入。
• 事件驱动：所有策略逻辑由事件触发（on_bar、on_tick、on_order_status），无需轮询。
• 先订阅再交易：在 init() 中使用 subscribe() 确保 on_bar/on_tick 触发前数据已就绪。
• 回测模式init中不可下单：回测模式下 init() 仅允许订阅和参数设置。
• 定期调仓用schedule：使用 schedule() 而非 on_bar 实现定期调仓策略。
• df=True：大多数数据查询函数支持 df=True 返回pandas DataFrame，便于分析。
• 多账户：在下单函数中传入 account='' 参数指定账户。
• 期货平仓：上期所需要用 PositionEffect_CloseToday / PositionEffect_CloseYesterday 区分平今仓和平昨仓。
• 回测默认值：初始资金=1,000,000，手续费=0，滑点=0。请设置合理值。
• Documentation: https://www.myquant.cn/docs/python/41

进阶示例

多因子选股策略

from gm.api import *
import numpy as np

def init(context):
    # Execute stock selection and rebalancing at 10:00 on the 1st trading day of each month
    schedule(schedule_func=multi_factor_rebalance, date_rule='1m', time_rule='10:00:00')
    context.hold_num = 10          # 持股数量
    context.target_index = 'SHSE.000300'  # Selection universe: CSI 300 constituents

def multi_factor_rebalance(context):
    """Multi-factor stock selection + equal-weight rebalancing"""
    # 获取沪深300成分股
    constituents = get_constituents(index=context.target_index)
    symbols = [c['symbol'] for c in constituents]

    # Get financial indicator data (PE, PB, ROE, turnover rate)
    fund_data = get_fundamentals(
        table='trading_derivative_indicator',
        symbols=','.join(symbols),
        end_date=context.now.strftime('%Y-%m-%d'),
        count=1,
        fields='PETTM,PB,TURNRATE',
        df=True
    )

    if fund_data is None or len(fund_data) == 0:
        return

    # Data cleaning: remove missing values and outliers
    fund_data = fund_data.dropna(subset=['PETTM', 'PB'])
    fund_data = fund_data[(fund_data['PETTM'] > 0) & (fund_data['PETTM'] < 100)]
    fund_data = fund_data[(fund_data['PB'] > 0) & (fund_data['PB'] < 20)]

    # Factor scoring: lower PE is better, lower PB is better (higher rank = higher score)
    fund_data['PE_rank'] = fund_data['PETTM'].rank(ascending=True)
    fund_data['PB_rank'] = fund_data['PB'].rank(ascending=True)
    # Composite score = PE rank * 0.5 + PB rank * 0.5
    fund_data['score'] = fund_data['PE_rank'] * 0.5 + fund_data['PB_rank'] * 0.5
    # Sort by composite score, select top N
    selected = fund_data.nsmallest(context.hold_num, 'score')
    target_symbols = selected['symbol'].tolist()

    print(f"Selected {len(target_symbols)} stocks this period: {target_symbols}")

    # First close positions not in the target list
    positions = context.account().positions()
    for pos in positions:
        if pos['symbol'] not in target_symbols and pos['volume'] > 0:
            order_target_volume(symbol=pos['symbol'], volume=0,
                                position_side=PositionSide_Long,
                                order_type=OrderType_Market)
            print(f"  Closing: {pos['symbol']}")

    # 等权重买入目标股票
    target_pct = 0.95 / len(target_symbols)  # Keep 5% cash
    for sym in target_symbols:
        order_target_percent(symbol=sym, percent=target_pct,
                             position_side=PositionSide_Long,
                             order_type=OrderType_Market)
        print(f"  Rebalancing: {sym} -> {target_pct*100:.1f}%")

if __name__ == '__main__':
    run(strategy_id='multi_factor',
        filename='main.py',
        mode=MODE_BACKTEST,
        token='your_token',
        backtest_start_time='2023-01-01 09:30:00',
        backtest_end_time='2024-06-30 15:00:00',
        backtest_initial_cash=1000000,
        backtest_commission_ratio=0.0003,
        backtest_slippage_ratio=0.001,
        backtest_adjust=ADJUST_PREV)

配对交易策略（统计套利）

from gm.api import *
import numpy as np

def init(context):
    # Pair: China Merchants Bank vs Industrial Bank (same industry, high correlation)
    context.stock_a = 'SHSE.600036'
    context.stock_b = 'SHSE.601166'
    context.lookback = 60          # Lookback window (trading days)
    context.entry_z = 2.0          # Entry Z-score threshold
    context.exit_z = 0.5           # Exit Z-score threshold
    context.position_pct = 0.4     # Single-side position ratio

    # Subscribe to daily bars for both stocks
    subscribe(symbols=f'{context.stock_a},{context.stock_b}', frequency='1d', count=context.lookback + 1)

def on_bar(context, bars):
    # Get historical close prices
    hist_a = context.data(symbol=context.stock_a, frequency='1d', count=context.lookback)
    hist_b = context.data(symbol=context.stock_b, frequency='1d', count=context.lookback)

    if len(hist_a) < context.lookback or len(hist_b) < context.lookback:
        return

    closes_a = np.array([bar['close'] for bar in hist_a])
    closes_b = np.array([bar['close'] for bar in hist_b])

    # Calculate price ratio (spread)
    ratio = closes_a / closes_b
    ratio_mean = np.mean(ratio)
    ratio_std = np.std(ratio)

    if ratio_std == 0:
        return

    # Calculate current Z-score
    current_ratio = bars[0]['close'] / bars[1]['close'] if len(bars) >= 2 else ratio[-1]
    z_score = (current_ratio - ratio_mean) / ratio_std

    # Query current positions
    pos_a = context.account().position(symbol=context.stock_a, side=PositionSide_Long)
    pos_b = context.account().position(symbol=context.stock_b, side=PositionSide_Long)
    has_pos_a = pos_a is not None and pos_a['volume'] > 0
    has_pos_b = pos_b is not None and pos_b['volume'] > 0

    print(f"Z-score={z_score:.2f}, ratio={current_ratio:.4f}, mean={ratio_mean:.4f}")

    if z_score > context.entry_z and not has_pos_b:
        # Z-score too high: A overvalued relative to B → sell A, buy B
        if has_pos_a:
            order_close_all()  # Close reverse positions first
        order_percent(symbol=context.stock_b, percent=context.position_pct,
                      side=OrderSide_Buy, order_type=OrderType_Market,
                      position_effect=PositionEffect_Open)
        print(f"  Open position: buy {context.stock_b}")

    elif z_score < -context.entry_z and not has_pos_a:
        # Z-score too low: A undervalued relative to B → buy A, sell B
        if has_pos_b:
            order_close_all()
        order_percent(symbol=context.stock_a, percent=context.position_pct,
                      side=OrderSide_Buy, order_type=OrderType_Market,
                      position_effect=PositionEffect_Open)
        print(f"  Open position: buy {context.stock_a}")

    elif abs(z_score) < context.exit_z and (has_pos_a or has_pos_b):
        # Z-score reverts to mean → close positions
        order_close_all()
        print(f"  Close positions: Z-score reverted")

if __name__ == '__main__':
    run(strategy_id='pair_trading',
        filename='main.py',
        mode=MODE_BACKTEST,
        token='your_token',
        backtest_start_time='2023-01-01 09:30:00',
        backtest_end_time='2024-06-30 15:00:00',
        backtest_initial_cash=1000000,
        backtest_commission_ratio=0.0003,
        backtest_adjust=ADJUST_PREV)

期货CTA趋势跟踪策略（海龟交易）

from gm.api import *
import numpy as np

def init(context):
    context.symbol = 'CFFEX.IF2401'   # CSI 300 index futures
    context.entry_period = 20          # 入场通道周期 (20-day high/low)
    context.exit_period = 10           # 出场通道周期 (10-day high/low)
    context.atr_period = 20            # ATR calculation period
    context.risk_ratio = 0.01          # Per-trade risk ratio (1% of total assets)

    subscribe(symbols=context.symbol, frequency='1d', count=context.entry_period + 1)

def on_bar(context, bars):
    hist = context.data(symbol=context.symbol, frequency='1d', count=context.entry_period)
    if len(hist) < context.entry_period:
        return

    highs = np.array([bar['high'] for bar in hist])
    lows = np.array([bar['low'] for bar in hist])
    closes = np.array([bar['close'] for bar in hist])

    # Calculate Donchian Channel
    entry_high = np.max(highs[-context.entry_period:])    # 20-day high (breakout long)
    entry_low = np.min(lows[-context.entry_period:])      # 20-day low (breakout short)
    exit_high = np.max(highs[-context.exit_period:])      # 10-day high (short stop-loss)
    exit_low = np.min(lows[-context.exit_period:])        # 10-day low (long stop-loss)

    # Calculate ATR (Average True Range)
    tr_list = []
    for i in range(1, len(highs)):
        tr = max(highs[i] - lows[i],
                 abs(highs[i] - closes[i-1]),
                 abs(lows[i] - closes[i-1]))
        tr_list.append(tr)
    atr = np.mean(tr_list[-context.atr_period:]) if len(tr_list) >= context.atr_period else 0

    current_price = bars[0]['close']
    pos_long = context.account().position(symbol=context.symbol, side=PositionSide_Long)
    pos_short = context.account().position(symbol=context.symbol, side=PositionSide_Short)
    has_long = pos_long is not None and pos_long['volume'] > 0
    has_short = pos_short is not None and pos_short['volume'] > 0

    # 计算仓位大小（基于ATR的风险管理）
    if atr > 0:
        nav = context.account().cash['nav']
        unit_size = int(nav * context.risk_ratio / (atr * 300))  # IF contract multiplier 300
        unit_size = max(unit_size, 1)
    else:
        unit_size = 1

    # Entry signals
    if current_price > entry_high and not has_long:
        if has_short:
            # Close short position first
            order_volume(symbol=context.symbol, volume=pos_short['volume'],
                         side=OrderSide_Buy, order_type=OrderType_Market,
                         position_effect=PositionEffect_Close)
        # Open long position
        order_volume(symbol=context.symbol, volume=unit_size,
                     side=OrderSide_Buy, order_type=OrderType_Market,
                     position_effect=PositionEffect_Open)
        print(f"Breakout long: price={current_price}, upper channel={entry_high}, lots={unit_size}")

    elif current_price < entry_low and not has_short:
        if has_long:
            order_volume(symbol=context.symbol, volume=pos_long['volume'],
                         side=OrderSide_Sell, order_type=OrderType_Market,
                         position_effect=PositionEffect_Close)
        # Open short position
        order_volume(symbol=context.symbol, volume=unit_size,
                     side=OrderSide_Sell, order_type=OrderType_Market,
                     position_effect=PositionEffect_Open)
        print(f"Breakout short: price={current_price}, lower channel={entry_low}, lots={unit_size}")

    # Exit signals
    elif has_long and current_price < exit_low:
        order_volume(symbol=context.symbol, volume=pos_long['volume'],
                     side=OrderSide_Sell, order_type=OrderType_Market,
                     position_effect=PositionEffect_Close)
        print(f"Long stop-loss: price={current_price}, exit lower channel={exit_low}")

    elif has_short and current_price > exit_high:
        order_volume(symbol=context.symbol, volume=pos_short['volume'],
                     side=OrderSide_Buy, order_type=OrderType_Market,
                     position_effect=PositionEffect_Close)
        print(f"Short stop-loss: price={current_price}, exit upper channel={exit_high}")

def on_backtest_finished(context, indicator):
    print(f"\nBacktest results:")
    print(f"  Return: {indicator['pnl_ratio']:.2%}")
    print(f"  Annualized return: {indicator['pnl_ratio_annual']:.2%}")
    print(f"  Sharpe ratio: {indicator['sharpe_ratio']:.2f}")
    print(f"  Max drawdown: {indicator['max_drawdown']:.2%}")
    print(f"  Win rate: {indicator['win_ratio']:.2%}")

if __name__ == '__main__':
    run(strategy_id='turtle_cta',
        filename='main.py',
        mode=MODE_BACKTEST,
        token='your_token',
        backtest_start_time='2023-01-01 09:30:00',
        backtest_end_time='2024-06-30 15:00:00',
        backtest_initial_cash=2000000,
        backtest_commission_ratio=0.000023,
        backtest_slippage_ratio=0.0005,
        backtest_adjust=ADJUST_PREV)

风险管理模块 — 仓位控制与回撤监控

from gm.api import *

def init(context):
    subscribe(symbols='SHSE.600000,SZSE.000001,SHSE.601318', frequency='1d', count=21)
    context.max_drawdown_limit = 0.10   # Max drawdown limit 10%
    context.max_single_position = 0.30  # Max single stock position 30%
    context.max_total_position = 0.90   # Max total position 90%
    context.peak_nav = 0                # Historical peak NAV
    context.is_stopped = False          # Whether risk control has stopped trading

def on_bar(context, bars):
    # Update historical peak NAV
    nav = context.account().cash['nav']
    if nav > context.peak_nav:
        context.peak_nav = nav

    # Calculate current drawdown
    current_drawdown = (context.peak_nav - nav) / context.peak_nav if context.peak_nav > 0 else 0

    # Check if max drawdown limit is triggered
    if current_drawdown >= context.max_drawdown_limit:
        if not context.is_stopped:
            print(f"⚠️ Max drawdown limit triggered: drawdown={current_drawdown:.2%}, closing all positions!")
            order_close_all()
            context.is_stopped = True
        return

    context.is_stopped = False

    # Check if any single stock position exceeds the limit
    positions = context.account().positions()
    total_position_value = sum(p['market_value'] for p in positions)
    total_asset = context.account().cash['nav']

    for pos in positions:
        single_pct = pos['market_value'] / total_asset if total_asset > 0 else 0
        if single_pct > context.max_single_position:
            # Single stock position exceeds limit, reduce to the limit
            target_value = total_asset * context.max_single_position
            order_target_value(symbol=pos['symbol'], value=target_value,
                               position_side=PositionSide_Long,
                               order_type=OrderType_Market)
            print(f"  Reducing: {pos['symbol']} position {single_pct:.1%} -> {context.max_single_position:.1%}")

    # Output risk control status
    total_pct = total_position_value / total_asset if total_asset > 0 else 0
    print(f"NAV={nav:.2f}, drawdown={current_drawdown:.2%}, total position={total_pct:.1%}")

if __name__ == '__main__':
    run(strategy_id='risk_mgmt',
        filename='main.py',
        mode=MODE_BACKTEST,
        token='your_token',
        backtest_start_time='2023-01-01 09:30:00',
        backtest_end_time='2024-06-30 15:00:00',
        backtest_initial_cash=1000000,
        backtest_adjust=ADJUST_PREV)

数据研究 — 行业轮动分析

from gm.api import *
import pandas as pd

# Set Token (pure data research, no strategy needed)
set_token('your_token')

# Define industry ETF list (using ETFs instead of industry indices for easier data access)
industry_etfs = {
    'SHSE.510050': '上证50',
    'SHSE.510300': '沪深300',
    'SHSE.510500': '中证500',
    'SZSE.159915': '创业板',
    'SHSE.512010': '医药ETF',
    'SHSE.512880': '证券ETF',
    'SHSE.512800': '银行ETF',
    'SHSE.515030': '新能源车ETF',
    'SZSE.159995': '芯片ETF',
    'SHSE.512690': '白酒ETF',
}

# Get daily bar data for each industry ETF over the past year
results = []
for symbol, name in industry_etfs.items():
    df = history(symbol=symbol, frequency='1d',
                 start_time='2024-01-01', end_time='2024-12-31',
                 fields='close,volume,eob',
                 adjust=ADJUST_PREV, df=True)
    if df is not None and len(df) > 20:
        # Calculate returns over various periods
        ret_5d = (df['close'].iloc[-1] / df['close'].iloc[-6] - 1) * 100    # 5-day return
        ret_20d = (df['close'].iloc[-1] / df['close'].iloc[-21] - 1) * 100  # 20-day return
        ret_60d = (df['close'].iloc[-1] / df['close'].iloc[-61] - 1) * 100 if len(df) > 60 else None
        avg_volume = df['volume'].tail(20).mean()  # 20-day average volume

        results.append({
            '行业': name,
            '代码': symbol,
            '近5日收益(%)': round(ret_5d, 2),
            '近20日收益(%)': round(ret_20d, 2),
            '近60日收益(%)': round(ret_60d, 2) if ret_60d else None,
            '20日均量': int(avg_volume),
        })

# Sort by 20-day return
df_result = pd.DataFrame(results).sort_values('近20日收益(%)', ascending=False)
print("\nIndustry rotation analysis (sorted by 20-day return):")
print(df_result.to_string(index=False))

# Find industries with strongest momentum (ranked top 3 in both 5-day and 20-day returns)
top5d = set(df_result.nlargest(3, '近5日收益(%)')['行业'].tolist())
top20d = set(df_result.nlargest(3, '近20日收益(%)')['行业'].tolist())
momentum_leaders = top5d & top20d
if momentum_leaders:
    print(f"\nMomentum leaders (strong in both short-term and mid-term): {momentum_leaders}")

🤖 AI Agent 高阶使用指南

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

1. 数据校验与错误处理

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

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

2. 多步组合分析

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

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

3. 构建动态监控与日志

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

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

社区与支持

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

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