2、简易量化交易策略实现

约 3046 字大约 10 分钟

2026-05-31

数据样本说明

股价常用指标实现

量化策略编写

(1) 核心策略

1、策略框架

框架: 初始化+策略函数（周期循环）

初始化：

通过初始化函数设置基准。初始化函数在整个回测或者实盘操作过程中只被运行一次，用于初始化全局变量
函数：initialize(context)
context: Context对象，存放有当前的账号/股票持仓信息

示例：

def initialize(context):

g为全局变量，设定标的股票为深交所的平安银行

g.security = "000001.XSHE"

策略函数：

策略开始后，随着时间周期重复执行你的交易策略。

def handle_data(context, data):

下单一千股

order(g.security, 1000)

卖出八百股股:

order(g.security, -800)

2、实用策略示例

初始化：

设定投资标的（平安银行）
设定策略运行周期（每日）

策略函数：

如果上一时间点价格高出五天平均价1%，则全仓买入
如果上一时间点价格低于五天平均价，则空仓卖出

(画出上一时间点价格)

策略编写:

# 初始化函数：回测开始前只执行一次
def initialize(context):
    # 全局变量：指定要交易的股票（平安银行）
    g.security = '000001.XSHE'
    
    # 运行策略函数：每根K线触发一次（这里是日线，每天执行一次）
    run_daily(market_open, time='every_bar')

# 策略核心逻辑：每天自动执行一次
def market_open(context):
    # 取出要交易的股票代码
    security = g.security
    
    # 获取历史数据：最近5天的日线收盘价
    # 参数：股票代码, 天数, 数据周期, 需要获取的字段（这里只取收盘价）
    close_data = attribute_history(security, 5, '1d', ['close'])
    
    # 计算5日均线：把最近5天收盘价求平均值
    MA5 = close_data['close'].mean()
    
    # 获取当前价格：取最近一天（最后一行）的收盘价
    current_price = close_data['close'][-1]

    # 获取账户当前可用现金（能用来买股票的钱）
    cash = context.portfolio.available_cash
    
    # ===================== 交易规则 =====================
    # 买入条件：当前价格 > 1.01倍5日均线（向上突破）
    if current_price > 1.01 * MA5:
        # 用所有可用现金全仓买入这只股票
        order_value(security, cash)
        # 打印日志：记录买入操作
        log.info("买入 %s" % (security))
    
    # 卖出条件：当前价格 < 5日均线 并且 持有股票可卖出数量 > 0（有持仓才卖）
    elif current_price < MA5 and context.portfolio.positions[security].closeable_amount > 0:
        # 卖出全部持仓，目标持仓数量=0
        order_target(security, 0)
        # 打印日志：记录卖出操作
        log.info("卖出 %s" % (security))
    
    # 绘图：把当前股价记录到回测图表
    record(stock_price=current_price)

(2) 策略函数设置

1、常用策略设置函数

基准：

设置业绩比较基准
set_benchmark(security)

佣金/印花税：

股票类每笔交易时的手续费为:买入时佣金万分之三，卖出时佣金万分之三加千分之一印花税
set_order_cost(cost, type, ref=None)

滑点：

真实的成交价格与下单时预期的价格的偏差
set_slippage(object,type=None, ref=None)

成交量比例：

根据实际行情限制每个订单的成交量
set_option('order_volume_ratio', value)

动态复权模式：

设置真实价格，建议开启
set_option('use_real_price', value)

2、代码案例

# 初始化函数：回测开始前，只执行一次
def initialize(context):
    # 1. 设置策略业绩基准（对比曲线）：沪深300指数
    set_benchmark("000300.XSHG")
    
    # 2. 全局变量：指定交易股票为 平安银行(000001)
    g.security = '000001.XSHE'
    
    # 3. 设置每日运行时间：每天上午10:00自动执行 market_open 函数
    run_daily(market_open, time='10:00')
    
    # 4. 设置股票交易手续费（真实模拟交易成本）
    set_order_cost(
        OrderCost(
            open_commission=0.03,      # 买入佣金：万分之3（0.03%）
            close_commission=0.03,     # 卖出佣金：万分之3（0.03%）
            close_tax=0.01,            # 卖出印花税：千分之1（0.1%）
            min_commission=5           # 单笔佣金最低收费5元
        ), 
        type='stock'  # 手续费规则适用于股票
    )
    
    # 5. 滑点设置（已注释，不生效）：设置0.2%的价格相关滑点
    #set_slippage(PriceRelatedSlippage(0.002),type='stock')
    
    # 6. 订单比例设置（已注释，不生效）
    #set_option('order_volume_ratio',0.5)
    
    # 7. 关闭“使用真实价格”：回测使用回测系统默认价格撮合
    set_option('use_real_price', False)

# ==================== 每日交易逻辑 ====================
# 每天 10:00 执行一次
def market_open(context):
    # 判断：如果当前账户 没有持有 平安银行
    if g.security not in context.portfolio.positions:
        # 买入 1000 股 平安银行
        order(g.security, 1000)
    
    # 如果账户 已经持有 平安银行
    else:
        # 卖出 800 股 平安银行
        order(g.security, -800)

(3) 定时函数

1、定时函数介绍

什么是定时函数?

定时函数参数:

定时函数：

设定回测和模拟交易中运行时间及频率

月度：

run_monthly(func,monthday,time='open',reference_secuity)

周度：

run_weekly(func,weekday,time='open',reference_secuity)

日度：

run_daily(func,time='open',reference_secuity)

2、案例

def initialize(context):
    # 设定业绩标的为“沪深300”
    set_benchmark("000300.XSHG")
    g.security = '000001.XSHE'
    # 定时交易
    #run_daily(market_open, time='10:00', reference_security='000300.XSHG')
    # 每分钟
    #run_daily(market_open, time='every_bar', reference_security='000300.XSHG')
    # 每月第一个交易日
    #run_monthly(market_open, 1, time='open')
    # 每周最后一个交易日
    run_weekly(market_open, -1, time='open')

def market_open(context):
    if g.security not in context.portfolio.positions:
        order(g.security, 1000)
    else:
        order(g.security, -800)

(4) 交易函数

1、交易数量

函数签名:order(security, amount, style=None, side='long', pindex=0)

security: 股票代码
amount: 交易数量（负数表示卖出）
style: 下单类型
side: short空（一般不允许）/ long多
pindex: 仓位号，默认为0

示例：

按限单价30买入100股

order('600000.XSHG', 100, LimitOrderStyle(30.0))

2、股票价值

order(security, value, style=None, side='long', pindex=0)

security: 股票代码

value: 股票价值（负数表示卖出）

style: 下单类型

side: short空（一般不允许）/ long多

pindex: 仓位号，默认为0

示例：

按卖出价值为5000元的股票

order('600000.XSHG', -5000)

3、目标数量

order_target(security, amount, style=None, side='long', pindex=0, close_today=False)

security: 股票代码

amount: 交易数量（负数表示卖出）

style: 下单类型

side: short空（一般不允许）/ long多

pindex: 仓位号，默认为0

示例：

买入平安银行所有股票到100股

order_target('000001.XSHE', 100)

4、查询成交订单

get_orders(order id=None,security=None,status=None)

示例：

get_orders(order_id='123')订单id查询订单号为"123"的订单

get_orders(security='000001.XSHE') 查询所有标的为000001.XSHE的订单

5、未完成订单

get_open_orders()

示例：

在每天交易结束之后获取当天所有的未完成订单

def after_market_close(context):

orders = get_open_orders(）

for order in orders:

log.info(_order.order_id)

6、撤单函数

cancel_order(order)

示例：

在每天交易结束之后对未完成订单进行撤单

def after_market_close(context):

orders = get_open_orders()

for _order in orders.values():

cancel_order(order)

7、账号出入金

inout_cash(cash, pindex=0)

cash: 浮点数，负数表示出金

pindex: 仓位号，默认为0

示例：

向账户增加10000元:

inout_cash(10000, pindex=0)

8、代码实战

def initialize(context):
    # 设定业绩标的为“沪深300”
    set_benchmark("000300.XSHG")
    g.security = '000001.XSHE'
    run_daily(market_open, time='9:30')
    run_daily(after_market_close, time='15:30')

def market_open(context):
    # 向账户增加10000元
    inout_cash(10000, pindex=0)
    # 查询可以资金
    log.info("""账户可用资金：{}""".format(context.portfolio.subportfolios[0].available_cash))
    
    # 获取账户当前现金
    cash = context.portfolio.available_cash
    
    #如果没有持仓
    if g.security not in context.portfolio.positions:
        #下单1000股
        order(g.security, 1000)
    else:
        #卖出500股
        order(g.security, -500)
        
        
def after_market_close(context):
    # 获取当天所有未完成的订单
    orders = get_open_orders()
    for _order in orders:
        log.info("""未完成订单：{}""".format(_order))
    # 对未完成订单进行撤单
    for _order in orders:
        cancel_order(_order)

(5) 交易对象

1、Order对象

Order对象：订单处理流程

订单创建 ->订单检查 -> 报单 -> 确认委托 -> 撮合

commission: 交易费用(佣金、税费等)
is_buy: bool值,买还是卖
status: 状态,一个OrderStatus值
price: 平均成交价格，已经成交的股票的平均成交价格

2、Trade对象

Trade对象：订单成交相关信息

time：交易时间，[datetime.datetime]对象
security：标的代码
amount：交易数量
price：交易价格
trade_id：交易记录id
order_id：对应的订单id

3、代码案例

# 初始化函数：回测开始前只执行一次
def initialize(context):
    # 设置业绩基准：沪深300指数
    set_benchmark("000300.XSHG")
    # 指定交易股票：平安银行(000001)
    g.security = '000001.XSHE'
    
    # 每天开盘自动执行交易函数
    run_daily(market_open, time='open')
    # 每天15:30收盘后执行日志记录函数
    run_daily(after_market_close, time='15:30')

# 开盘交易逻辑：每天开盘自动运行
def market_open(context):
    # 如果没有持仓 → 买入1000股
    if g.security not in context.portfolio.positions:
        order(g.security, 1000)
    # 如果有持仓 → 卖出800股
    else:
        order(g.security, -800)

# 收盘后函数：每天15:30自动运行
def after_market_close(context):
    print("闭市后，开始查询成交记录...")
    
    # 获取当天所有成交记录（字典格式）
    trades = get_trades()
    
    # 遍历所有成交记录
    for _trade in trades.values():
        # 打印成交详情
        print("成交记录:{}".format(_trade))
        print("成交时间:{}".format(_trade.time))
        print("对应的订单id:{}".format(_trade.order_id))


"""
def market_open(context):
    #如果没有持仓
    if g.security not in context.portfolio.positions:
        orders = order(g.security, 100)
        print(orders)
        
        if orders is None:
            print("创建订单失败...")
        else:
            print('''交易费用：{}'''.format(orders.commission))
            print('''是否买单：{}'''.format(orders.is_buy))
            print('''订单状态：{}'''.format(orders.status))
            print('''订单评价成交价格：{}'''.format(orders.price))
    else:
        #卖出500股
        order(g.security, -800)
"""

(6) 策略信息

1、Context对象

Context对象：策略信息总览，包含账户、时间等信息
subportfolios：当前单个操作仓位的资金、标的信息，是一个SubPortfolio的数组
portfolio：账户信息，即subportfolios 的汇总信息，Portfolio对象，单个操作仓位时，portfolio指向 subportfolios[0]
current_dt：当前单位时间的开始时间，[datetime.datetime]对象
previous date：datetime前一个交易日，[datetime.date]对象,注意，这是一个日期，是 date,而不是datetime
universe：查询set_universe()设定的股票池，比如['000001.XSHE','600000.XSHG']

2、Position对象

Position对象：输出持有的标的的信息

security：标的代码
price：最新行情价格
total_amount：总仓位，不包括挂单冻结仓位
init_time：建仓时间,格式为datetime.datetime

3、代码案例

def initialize(context):
    g.security = "000001.XSHE"
    
def handle_data(context, data):
    #如果没有持仓
    if g.security not in context.portfolio.positions:
        #下单1000股
        order(g.security, 1000)
    else:
        #卖出800股
        order(g.security, -800)
    print(type(context.portfolio.long_positions))
    long_positions_dict = context.portfolio.long_positions
    for position in list(long_positions_dict.values()):
        print(
            """"标的:{}，总仓位:{}，标的价值:{}，建仓时间:{}"""
            .format(position.security,
                position.total_amount,
                position.value,
                position.init_time)
        )
    

def handle_data(context, data):
    #context.portfolio变为整数1
    context.portfolio = 1
    log.info("context.portfolio")
    log.info(context.portfolio)
    
    #恢复系统变量
    del context.portfolio
    
    #context.portfolio将变成用户账户信息
    log.info("context.portfolio.total_value")
    log.info(context.portfolio.total_value)
    
    # 输出账户总资产
    log.info("context.portfolio.total_value")
    log.info(context.portfolio.total_value)
    
    # 输出持仓金额
    log.info("context.portfolio.positions_value")
    log.info(context.portfolio.positions_value)
    
    # 输出今日日期
    log.info("context.current_dt.day")
    log.info(context.current_dt.day)
    
    # 输出总权益的累计收益
    log.info("context.portfolio.returns")
    log.info(context.portfolio.returns)
    
    # 获取仓位subportfolios[0]的可用资金
    log.info("context.subportfolios[0].available_cash")
    log.info(context.subportfolios[0].available_cash)

7、账户信息

1、Portfolio对象

Portfolio对象：总账户信息

long_positions：

多单的仓位,一个dict, key是证券代码, value 是[Position]对象

short_positions：

空单的仓位,一个dict, key是证券代码, value 是[Position]对象

total_value：

总的权益，包括现金，保证金(期货)或者仓位(股票)的总价值，可用来计算收益

returns：

总权益的累计收益;(当前总资产+今日出入金-昨日总资产)/昨日总资产

starting_cash：

初始资金,现在等于inout_cash

positions_value：

持仓价值

2、SubPortfolio对象

SubPortfolio对象：子账户信息

inout_cash：

累计出入金,如初始资金1000,后来转移出去100,则这个值是1000 -100

available_cash：

可用资金,可用来购买证券的资金

transferable_cash：

可取资金，即可以提现的资金,不包括今日卖出证券所得资金

locked_cash：

挂单锁住资金

type：

账户所属类型

3、代码案例

def initialize(context):
    g.security = "000001.XSHE"
    
def handle_data(context, data):
    #如果没有持仓
    if g.security not in context.portfolio.positions:
        #下单1000股
        order(g.security, 1000)
    else:
        #卖出800股
        order(g.security, -800)
    
    print("""累计出入金:{}""".format(context.subportfolios[0].inout_cash))
    print("""可以资金:{}""".format(context.subportfolios[0].available_cash))
    print("""可取资金:{}""".format(context.subportfolios[0].transferable_cash))
    print("""挂单锁住资金:{}""".format(context.subportfolios[0].locked_cash))
    print("""账户所属类型:{}""".format(context.subportfolios[0].type))

'''
    print("""多单的仓位:{}""".format(context.portfolio.long_positions))
    print("""空单的仓位:{}""".format(context.portfolio.short_positions))
    print("""总权益:{}""".format(context.portfolio.total_value))
    print("""总权益的累计收益率:{}""".format(context.portfolio.returns))
    print("""初始资金:{}""".format(context.portfolio.starting_cash))
    print("""持仓价值:{}""".format(context.portfolio.positions_value))
'''

贡献者

rzfgentlevip

cmake

mpc

make

2、简易量化交易策略实现

数据样本说明

股价常用指标实现

量化策略编写

(1) 核心策略

1、策略框架

2、实用策略示例

(2) 策略函数设置

1、常用策略设置函数

2、代码案例

(3) 定时函数

1、定时函数介绍

2、案例

(4) 交易函数

1、交易数量

2、股票价值

3、目标数量

4、查询成交订单

5、未完成订单

6、撤单函数

7、账号出入金

8、代码实战

(5) 交易对象

1、Order对象

2、Trade对象

3、代码案例

(6) 策略信息

1、Context对象

2、Position对象

3、代码案例

7、账户信息

1、Portfolio对象

2、SubPortfolio对象

3、代码案例

贡献者